Split daily stats queries

.github/workflows/ci-build-upload-binaries.yml

@@ -21,7 +21,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ arc-linux-latest ]
+        platform: [ arc-ubuntu-22.04 ]
 
     runs-on: ${{ matrix.platform }}
     env:
@@ -38,14 +38,15 @@ jobs:
           rm -rf ci-builds || true
           mkdir -p $OUTPUT_DIR
           echo $OUTPUT_DIR
+
       - name: Install Dependencies (Linux)
         run: sudo apt-get update && sudo apt-get -y install libudev-dev
 
       - name: Sets env vars for tokio if set in manual dispatch inputs
-        if: github.event_name == 'workflow_dispatch' && inputs.add_tokio_unstable == true
         run: |
-          echo "RUSTFLAGS=--cfg tokio_unstable" >> $GITHUB_ENV
-          echo "CARGO_FEATURES=--features tokio-console" >> $GITHUB_ENV
+          echo 'RUSTFLAGS="--cfg tokio_unstable"' >> $GITHUB_ENV
+        if: github.event_name == 'workflow_dispatch' && inputs.add_tokio_unstable == true
+
       - name: Install Rust stable
         uses: actions-rs/toolchain@v1
         with:
@@ -102,6 +103,7 @@ jobs:
           if [ ${{ github.event_name == 'workflow_dispatch' && inputs.enable_deb == true }} = true ]; then
             cp target/debian/*.deb $OUTPUT_DIR
           fi
+
       - name: Deploy branch to CI www
         continue-on-error: true
         uses: easingthemes/ssh-deploy@main

.github/workflows/ci-build-vpn-api-wasm.yml

@@ -9,7 +9,7 @@ on:
 
 jobs:
   wasm:
-    runs-on: arc-linux-latest
+    runs-on: arc-ubuntu-22.04
     env:
       CARGO_TERM_COLOR: always
       RUSTUP_PERMIT_COPY_RENAME: 1

.github/workflows/ci-build.yml

@@ -38,7 +38,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ arc-linux-latest, custom-windows-11, custom-macos-15 ]
+        os: [ arc-ubuntu-22.04, custom-windows-11, custom-macos-15 ]
     runs-on: ${{ matrix.os }}
     env:
       CARGO_TERM_COLOR: always
@@ -46,9 +46,9 @@ jobs:
       RUSTUP_PERMIT_COPY_RENAME: 1
     steps:
       - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler cmake
+        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler
         continue-on-error: true
-        if: contains(matrix.os, 'linux')
+        if: contains(matrix.os, 'ubuntu')
 
       - name: Check out repository code
         uses: actions/checkout@v4
@@ -63,7 +63,7 @@ jobs:
 
       # To avoid running out of disk space, skip generating debug symbols
       - name: Set debug to false (unix)
-        if: contains(matrix.os, 'linux') || contains(matrix.os, 'mac')
+        if: contains(matrix.os, 'ubuntu') || contains(matrix.os, 'mac')
         run: |
           sed -i.bak 's/\[profile.dev\]/\[profile.dev\]\ndebug = false/' Cargo.toml
           git diff
@@ -93,14 +93,14 @@ jobs:
           command: build
 
       - name: Build all examples
-        if: contains(matrix.os, 'linux')
+        if: contains(matrix.os, 'ubuntu')
         uses: actions-rs/cargo@v1
         with:
           command: build
           args: --workspace --examples
 
       - name: Run all tests
-        if: contains(matrix.os, 'linux')
+        if: contains(matrix.os, 'ubuntu')
         uses: actions-rs/cargo@v1
         env:
           NYM_API: https://sandbox-nym-api1.nymtech.net/api
@@ -109,7 +109,7 @@ jobs:
           args: --workspace
 
       - name: Run expensive tests
-        if: (github.ref == 'refs/heads/develop' || github.event.pull_request.base.ref == 'develop' || github.event.pull_request.base.ref == 'master') && contains(matrix.os, 'linux')
+        if: (github.ref == 'refs/heads/develop' || github.event.pull_request.base.ref == 'develop' || github.event.pull_request.base.ref == 'master') && contains(matrix.os, 'ubuntu')
         uses: actions-rs/cargo@v1
         with:
           command: test

.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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/ci-nym-wallet-rust.yml

@@ -11,7 +11,7 @@ on:
 
 jobs:
   build:
-    runs-on: arc-linux-latest
+    runs-on: arc-ubuntu-22.04
     env:
       CARGO_TERM_COLOR: always
       RUSTUP_PERMIT_COPY_RENAME: 1

.github/workflows/ci-sdk-wasm.yml

@@ -11,7 +11,7 @@ on:
 
 jobs:
   wasm:
-    runs-on: arc-linux-latest
+    runs-on: arc-ubuntu-22.04
     env:
       CARGO_TERM_COLOR: always
       RUSTUP_PERMIT_COPY_RENAME: 1

.github/workflows/greetings.yml

@@ -6,7 +6,7 @@ jobs:
   greeting:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/first-interaction@v3
+    - uses: actions/first-interaction@v1
       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@v5
+        uses: actions/download-artifact@v4
         with:
           name: report
           path: .github/workflows/support-files/notifications

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

@@ -30,11 +30,13 @@ 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 }}
@@ -54,7 +56,7 @@ jobs:
       - name: Install Rust stable
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.88.0
+          toolchain: 1.86.0
           override: true
 
       - name: Build all binaries
@@ -74,6 +76,7 @@ jobs:
             target/release/nym-network-requester
             target/release/nym-cli
             target/release/nymvisor
+            target/release/nym-node
           retention-days: 30
     
       - id: create-release
@@ -88,6 +91,7 @@ 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@v5
+        uses: actions/setup-java@v4
         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@v5
+        uses: actions/download-artifact@v4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/nym-network-monitor/Cargo.toml

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

@@ -40,8 +40,7 @@ jobs:
       - name: Get version from cargo.toml
         id: get_version
         run: |
-          VERSION=$(yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml)
-          echo "result=$VERSION" >> $GITHUB_OUTPUT
+          yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml
 
       - name: cleanup-gateway-probe-ref
         id: cleanup_gateway_probe_ref
@@ -53,16 +52,13 @@ jobs:
       - name: Set GIT_TAG variable
         run: echo "GIT_TAG=${{ env.WORKING_DIRECTORY }}-${{ steps.get_version.outputs.result }}-${{ steps.cleanup_gateway_probe_ref.outputs.git_ref }}" >> $GITHUB_ENV
 
-      - name: Initialize RELEASE_TAG
-        run: echo "RELEASE_TAG=" >> $GITHUB_ENV
-      
-      - name: Set RELEASE_TAG for release
+      - name: Set RELEASE_TAG variable
         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 }}-${{ steps.cleanup_gateway_probe_ref.outputs.git_ref }}" >> $GITHUB_ENV
-        
+
       - name: New env vars
         run: echo "RELEASE_TAG='$RELEASE_TAG' GIT_TAG='$GIT_TAG' IMAGE_NAME_AND_TAGS='$IMAGE_NAME_AND_TAGS'"        
 

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

@@ -34,22 +34,18 @@ jobs:
       - name: Get version from cargo.toml
         id: get_version
         run: |
-          VERSION=$(yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml)
-          echo "result=$VERSION" >> $GITHUB_OUTPUT
+          yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml
 
       - name: Set GIT_TAG variable
         run: echo "GIT_TAG=${{ env.WORKING_DIRECTORY }}-${{ steps.get_version.outputs.result }}" >> $GITHUB_ENV
 
-      - name: Initialise RELEASE_TAG
-        run: echo "RELEASE_TAG=" >> $GITHUB_ENV
-      
-      - name: Set RELEASE_TAG for release
+      - name: Set RELEASE_TAG variable
         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'"        
 

.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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         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.47.1
+        uses: mikefarah/yq@v4.45.4
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

CHANGELOG.md

@@ -4,84 +4,6 @@ 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])
-- WG exit policy scripts update ([#5921])
-- Make DNS Resolver fallback optional ([#5920])
-- nym-node debug command to reset providers db ([#5914])
-- basic zulip client for sending messages ([#5913])
-- chore: allow compatibility with 'CDLA-Permissive-2.0' ([#5910])
-- feat: ecash liveness check ([#5890])
-- Remove old free credential handle ([#5864])
-
-[#5960]: https://github.com/nymtech/nym/pull/5960
-[#5921]: https://github.com/nymtech/nym/pull/5921
-[#5920]: https://github.com/nymtech/nym/pull/5920
-[#5914]: https://github.com/nymtech/nym/pull/5914
-[#5913]: https://github.com/nymtech/nym/pull/5913
-[#5910]: https://github.com/nymtech/nym/pull/5910
-[#5890]: https://github.com/nymtech/nym/pull/5890
-[#5864]: https://github.com/nymtech/nym/pull/5864
-
-## [2025.14-feta] (2025-08-05)
-
-- chore: nym node tokio console ([#5909])
-- Feature/dkg snapshot epoch ([#5900])
-- Feature/dkg epoch dealers query ([#5899])
-- sqlx-pool-guard: allocate more memory on windows ([#5896])
-- Support mnemonic in the NS agent ([#5883])
-- Allow PG database backend ([#5880])
-
-[#5909]: https://github.com/nymtech/nym/pull/5909
-[#5900]: https://github.com/nymtech/nym/pull/5900
-[#5899]: https://github.com/nymtech/nym/pull/5899
-[#5896]: https://github.com/nymtech/nym/pull/5896
-[#5883]: https://github.com/nymtech/nym/pull/5883
-[#5880]: https://github.com/nymtech/nym/pull/5880
-
 ## [2025.13-emmental] (2025-07-22)
 
 - fix: don't allow mixnode running in exit mode ([#5898])

Cargo.toml

@@ -39,11 +39,9 @@ members = [
     "common/cosmwasm-smart-contracts/ecash-contract",
     "common/cosmwasm-smart-contracts/group-contract",
     "common/cosmwasm-smart-contracts/mixnet-contract",
-    "common/cosmwasm-smart-contracts/multisig-contract",
-    "common/cosmwasm-smart-contracts/nym-performance-contract",
+    "common/cosmwasm-smart-contracts/multisig-contract", "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",
@@ -51,14 +49,13 @@ members = [
     "common/credentials-interface",
     "common/crypto",
     "common/dkg",
-    "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-macro",
+    "common/http-api-client",
     "common/http-api-common",
     "common/inclusion-probability",
     "common/ip-packet-requests",
@@ -92,32 +89,25 @@ members = [
     "common/socks5/requests",
     "common/statistics",
     "common/store-cipher",
-    "common/task", "common/test-utils",
+    "common/task",
     "common/ticketbooks-merkle",
     "common/topology",
     "common/tun",
-    "common/types", "common/upgrade-mode-check",
+    "common/types",
     "common/verloc",
     "common/wasm/client-core",
     "common/wasm/storage",
     "common/wasm/utils",
     "common/wireguard",
-    "common/wireguard-private-metadata/client",
-    "common/wireguard-private-metadata/server",
-    "common/wireguard-private-metadata/shared",
-    "common/wireguard-private-metadata/tests",
     "common/wireguard-types",
-    "common/zulip-client",
     "documentation/autodoc",
     "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",
@@ -125,15 +115,15 @@ members = [
     "nym-node-status-api/nym-node-status-client",
     "nym-node/nym-node-metrics",
     "nym-node/nym-node-requests",
-    "nym-outfox", "nym-signers-monitor",
+    "nym-outfox",
     "nym-statistics-api",
     "nym-validator-rewarder",
-    "nym-wg-gateway-client",
     "nyx-chain-watcher",
     "sdk/ffi/cpp",
     "sdk/ffi/go",
     "sdk/ffi/shared",
     "sdk/rust/nym-sdk",
+    "service-providers/authenticator",
     "service-providers/common",
     "service-providers/ip-packet-router",
     "service-providers/network-requester",
@@ -171,6 +161,7 @@ default-members = [
     "nym-statistics-api",
     "nym-validator-rewarder",
     "nyx-chain-watcher",
+    "service-providers/authenticator",
     "service-providers/ip-packet-router",
     "service-providers/network-requester",
     "tools/nymvisor",
@@ -185,7 +176,7 @@ homepage = "https://nymtech.net"
 documentation = "https://nymtech.net"
 edition = "2021"
 license = "Apache-2.0"
-rust-version = "1.81"
+rust-version = "1.80"
 readme = "README.md"
 
 [workspace.dependencies]
@@ -226,8 +217,8 @@ clap_complete = "4.5"
 clap_complete_fig = "4.5"
 colored = "2.2"
 comfy-table = "7.1.4"
-console = "0.16.0"
-console-subscriber = "0.4.1"
+console = "0.15.11"
+console-subscriber = "0.1.1"
 console_error_panic_hook = "0.1"
 const-str = "0.5.6"
 const_format = "0.2.34"
@@ -273,13 +264,11 @@ humantime = "2.2.0"
 humantime-serde = "1.1.1"
 hyper = "1.6.0"
 hyper-util = "0.1"
-indicatif = "0.18.0"
+indicatif = "0.17.11"
 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"
@@ -292,11 +281,8 @@ nix = "0.27.1"
 notify = "5.1.0"
 okapi = "0.7.0"
 once_cell = "1.21.3"
-opentelemetry = "0.30.0"
-opentelemetry-otlp = "0.30.0"
-opentelemetry-semantic-conventions = "0.30.0"
-opentelemetry_sdk = "0.30.0"
-opentelemetry-stdout = "0.30.0"
+opentelemetry = "0.19.0"
+opentelemetry-jaeger = "0.18.0"
 parking_lot = "0.12.3"
 pem = "0.8"
 petgraph = "0.6.5"
@@ -326,24 +312,22 @@ 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"
 sphinx-packet = "=0.6.0"
 sqlx = "0.8.6"
-strum = "0.27.2"
-strum_macros = "0.27.2"
+strum = "0.26"
+strum_macros = "0.26"
 subtle-encoding = "0.5"
-syn = "2"
-sysinfo = "0.37.0"
+syn = "1"
+sysinfo = "0.33.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.47"
+tokio = "1.45"
 tokio-postgres = "0.7"
 tokio-stream = "0.1.17"
 tokio-test = "0.4.4"
@@ -354,10 +338,8 @@ toml = "0.8.22"
 tower = "0.5.2"
 tower-http = "0.5.2"
 tracing = "0.1.41"
-tracing-core = "0.1.33"
 tracing-log = "0.2"
-tracing-opentelemetry = "0.31.0"
-tracing-serde = "0.2.0"
+tracing-opentelemetry = "0.19.0"
 tracing-subscriber = "0.3.19"
 tracing-tree = "0.2.2"
 tracing-indicatif = "0.3.9"
@@ -453,9 +435,6 @@ opt-level = 'z'
 # lto = true
 opt-level = 'z'
 
-[workspace.lints.rust]
-unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
-
 [workspace.lints.clippy]
 unwrap_used = "deny"
 expect_used = "deny"

clients/native/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-client"
-version = "1.1.63"
+version = "1.1.59"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 description = "Implementation of the Nym Client"
 edition = "2021"
@@ -46,6 +46,7 @@ nym-bandwidth-controller = { path = "../../common/bandwidth-controller" }
 nym-bin-common = { path = "../../common/bin-common", features = [
     "output_format",
     "clap",
+    "basic_tracing",
 ] }
 nym-client-core = { path = "../../common/client-core", features = [
     "fs-credentials-storage",

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::ShutdownManager;
+use nym_task::TaskHandle;
 use nym_validator_client::QueryHttpRpcNyxdClient;
 use std::error::Error;
 use std::path::PathBuf;
@@ -29,8 +29,6 @@ pub struct SocketClient {
 
     /// Optional path to a .json file containing standalone network details.
     custom_mixnet: Option<PathBuf>,
-
-    shutdown_manager: ShutdownManager,
 }
 
 impl SocketClient {
@@ -42,7 +40,6 @@ impl SocketClient {
         SocketClient {
             config,
             custom_mixnet,
-            shutdown_manager: Default::default(),
         }
     }
 
@@ -52,7 +49,7 @@ impl SocketClient {
         client_output: ClientOutput,
         client_state: ClientState,
         self_address: &Recipient,
-        shutdown_token: nym_task::ShutdownToken,
+        task_client: nym_task::TaskClient,
         packet_type: PacketType,
     ) {
         info!("Starting websocket listener...");
@@ -80,24 +77,24 @@ impl SocketClient {
             shared_lane_queue_lengths,
             reply_controller_sender,
             Some(packet_type),
-            shutdown_token.clone(),
+            task_client.fork("websocket_handler"),
         );
 
         websocket::Listener::new(
             config.socket.host,
             config.socket.listening_port,
-            shutdown_token.child_token(),
+            task_client.with_suffix("websocket_listener"),
         )
         .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 mut shutdown = self.start_socket().await?;
+        let shutdown = self.start_socket().await?;
 
-        shutdown.run_until_shutdown().await;
+        let res = shutdown.wait_for_shutdown().await;
         log::info!("Stopping nym-client");
-        Ok(())
+        res
     }
 
     async fn initialise_storage(&self) -> Result<OnDiskPersistent, ClientError> {
@@ -114,7 +111,7 @@ impl SocketClient {
         let dkg_query_client = if self.config.base.client.disabled_credentials_mode {
             None
         } else {
-            Some(default_query_dkg_client_from_config(&self.config.base)?)
+            Some(default_query_dkg_client_from_config(&self.config.base))
         };
 
         let storage = self.initialise_storage().await?;
@@ -122,7 +119,6 @@ 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 {
@@ -132,7 +128,7 @@ impl SocketClient {
         Ok(base_client)
     }
 
-    pub async fn start_socket(self) -> Result<ShutdownManager, ClientError> {
+    pub async fn start_socket(self) -> Result<TaskHandle, ClientError> {
         if !self.config.socket.socket_type.is_websocket() {
             return Err(ClientError::InvalidSocketMode);
         }
@@ -151,13 +147,13 @@ impl SocketClient {
             client_output,
             client_state,
             &self_address,
-            self.shutdown_manager.child_shutdown_token(),
+            started_client.task_handle.get_handle(),
             packet_type,
         );
 
         info!("Client startup finished!");
         info!("The address of this client is: {self_address}");
 
-        Ok(self.shutdown_manager)
+        Ok(started_client.task_handle)
     }
 }

clients/native/src/main.rs

@@ -4,7 +4,7 @@
 use std::error::Error;
 
 use clap::{crate_name, crate_version, Parser};
-use nym_bin_common::logging::{maybe_print_banner, setup_no_otel_logger};
+use nym_bin_common::logging::{maybe_print_banner, setup_tracing_logger};
 use nym_network_defaults::setup_env;
 
 pub mod client;
@@ -20,7 +20,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     if !args.no_banner {
         maybe_print_banner(crate_name!(), crate_version!());
     }
-    setup_no_otel_logger().expect("failed to initialize logging");
+    setup_tracing_logger();
 
     if let Err(err) = commands::execute(args).await {
         log::error!("{err}");

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::ShutdownToken;
+use nym_task::TaskClient;
 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>,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl HandlerBuilder {
@@ -57,7 +57,7 @@ impl HandlerBuilder {
         lane_queue_lengths: LaneQueueLengths,
         reply_controller_sender: ReplyControllerSender,
         packet_type: Option<PacketType>,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         Self {
             msg_input,
@@ -67,13 +67,14 @@ impl HandlerBuilder {
             lane_queue_lengths,
             reply_controller_sender,
             packet_type,
-            shutdown_token,
+            task_client,
         }
     }
 
     // TODO: make sure we only ever have one active handler
     pub fn create_active_handler(&self) -> Handler {
-        let shutdown_token = self.shutdown_token.clone();
+        let mut task_client = self.task_client.fork("active_handler");
+        task_client.disarm();
         Handler {
             msg_input: self.msg_input.clone(),
             client_connection_tx: self.client_connection_tx.clone(),
@@ -84,7 +85,7 @@ impl HandlerBuilder {
             lane_queue_lengths: self.lane_queue_lengths.clone(),
             reply_controller_sender: self.reply_controller_sender.clone(),
             packet_type: self.packet_type,
-            shutdown_token,
+            task_client,
         }
     }
 }
@@ -99,14 +100,19 @@ pub(crate) struct Handler {
     lane_queue_lengths: LaneQueueLengths,
     reply_controller_sender: ReplyControllerSender,
     packet_type: Option<PacketType>,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl Drop for Handler {
     fn drop(&mut self) {
-        let _ = self
+        if let Err(err) = self
             .buffer_requester
-            .unbounded_send(ReceivedBufferMessage::ReceiverDisconnect);
+            .unbounded_send(ReceivedBufferMessage::ReceiverDisconnect)
+        {
+            if !self.task_client.is_shutdown_poll() {
+                error!("failed to disconnect the receiver from the buffer: {err}");
+            }
+        }
     }
 }
 
@@ -136,7 +142,7 @@ impl Handler {
         {
             Ok(length) => length,
             Err(err) => {
-                if !self.shutdown_token.is_cancelled() {
+                if !self.task_client.is_shutdown_poll() {
                     error!(
                         "Failed to get reply queue length for connection {connection_id}: {err}"
                     );
@@ -184,9 +190,9 @@ impl Handler {
         });
 
         // the ack control is now responsible for chunking, etc.
-        let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type, None);
+        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.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send message to the input buffer: {err}");
             }
         }
@@ -217,9 +223,9 @@ impl Handler {
         });
 
         let input_msg =
-            InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type, None);
+            InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
         if let Err(err) = self.msg_input.send(input_msg).await {
-            if !self.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send anonymous message to the input buffer: {err}");
             }
         }
@@ -247,7 +253,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.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send reply message to the input buffer: {err}");
             }
         }
@@ -269,7 +275,7 @@ impl Handler {
             .client_connection_tx
             .unbounded_send(ConnectionCommand::Close(connection_id))
         {
-            if !self.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send close connection command: {err}");
             }
         }
@@ -388,14 +394,11 @@ impl Handler {
     }
 
     async fn listen_for_requests(&mut self, mut msg_receiver: ReconstructedMessagesReceiver) {
-        let shutdown_token = self.shutdown_token.clone();
+        let mut task_client = self.task_client.fork("select");
+        task_client.disarm();
 
-        loop {
+        while !task_client.is_shutdown() {
             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() {
@@ -433,6 +436,9 @@ impl Handler {
                         break;
                     }
                 }
+                _ = task_client.recv() => {
+                    log::trace!("Websocket handler: Received shutdown");
+                }
             }
         }
         log::debug!("Websocket handler: Exiting");
@@ -458,7 +464,7 @@ impl Handler {
                     reconstructed_sender,
                 ))
         {
-            if !self.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 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::ShutdownToken;
+use nym_task::TaskClient;
 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,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl Listener {
-    pub(crate) fn new(host: IpAddr, port: u16, shutdown_token: ShutdownToken) -> Self {
+    pub(crate) fn new(host: IpAddr, port: u16, task_client: TaskClient) -> Self {
         Listener {
             address: SocketAddr::new(host, port),
             state: State::AwaitingConnection,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -46,11 +46,11 @@ impl Listener {
 
         let notify = Arc::new(Notify::new());
 
-        while !self.shutdown_token.is_cancelled() {
+        while !self.task_client.is_shutdown() {
             tokio::select! {
                 // When the handler finishes we check if shutdown is signalled
                 _ = notify.notified() => {
-                    if self.shutdown_token.is_cancelled() {
+                    if self.task_client.is_shutdown() {
                         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.shutdown_token.cancelled() => {
+                _ = self.task_client.recv() => {
                     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.63"
+version = "1.1.59"
 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"
@@ -27,6 +27,7 @@ zeroize = { workspace = true }
 nym-bin-common = { path = "../../common/bin-common", features = [
     "output_format",
     "clap",
+    "basic_tracing",
 ] }
 nym-client-core = { path = "../../common/client-core", features = [
     "fs-credentials-storage",

clients/socks5/src/main.rs

@@ -4,7 +4,7 @@
 use std::error::Error;
 
 use clap::{crate_name, crate_version, Parser};
-use nym_bin_common::logging::{maybe_print_banner, setup_no_otel_logger};
+use nym_bin_common::logging::{maybe_print_banner, setup_tracing_logger};
 use nym_network_defaults::setup_env;
 
 mod commands;
@@ -19,7 +19,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     if !args.no_banner {
         maybe_print_banner(crate_name!(), crate_version!());
     }
-    setup_no_otel_logger().expect("failed to initialize logging");
+    setup_tracing_logger();
 
     if let Err(err) = commands::execute(args).await {
         log::error!("{err}");

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, NymApiClientExt};
+use nym_validator_client::nym_api::EpochId;
 use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
 use nym_validator_client::EcashApiClient;
 use rand::prelude::SliceRandom;
@@ -207,7 +207,7 @@ where
     <St as Storage>::StorageError: Send + Sync + 'static,
 {
     if let Some(stored) = storage
-        .get_expiration_date_signatures(expiration_date, epoch_id)
+        .get_expiration_date_signatures(expiration_date)
         .await
         .map_err(BandwidthControllerError::credential_storage_error)?
     {
@@ -220,7 +220,7 @@ where
         ecash_apis,
         |api| async move {
             api.api_client
-                .global_expiration_date_signatures(Some(expiration_date), Some(epoch_id))
+                .global_expiration_date_signatures(Some(expiration_date))
                 .await
         },
         format!("aggregated coin index signatures for date {expiration_date}"),

common/bin-common/Cargo.toml

@@ -8,30 +8,24 @@ license = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
-chrono = { workspace = true, optional = true }
-cfg-if = { workspace = true }
 clap = { workspace = true, features = ["derive"], optional = true }
 clap_complete = { workspace = true, optional = true }
 clap_complete_fig = { workspace = true, optional = true }
 const-str = { workspace = true }
-opentelemetry = { workspace = true, optional = true }
-opentelemetry-otlp = { workspace = true,features=["metrics", "grpc-tonic", "tls",
-    "tls-webpki-roots"], optional = true }
-opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"], optional = true }
-opentelemetry-stdout = { workspace = true, features = ["trace", "metrics"], optional = true }
-opentelemetry_sdk = { workspace = true, optional = true }
-rand = { workspace = true, optional = true }
+log = { workspace = true }
 schemars = { workspace = true, features = ["preserve_order"], optional = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, optional = true }
-thiserror = { workspace = true }
-tracing = { workspace = true }
-tracing-core = { workspace = true }
+
+## tracing
+tracing-subscriber = { workspace = true, features = ["env-filter"], optional = true }
+tracing-tree = { workspace = true, optional = true }
+tracing = { workspace = true, optional = true }
+opentelemetry-jaeger = { workspace = true, features = ["rt-tokio", "collector_client", "isahc_collector_client"], optional = true }
 tracing-opentelemetry = { workspace = true, optional = true }
-tracing-serde = { workspace = true }
-tracing-subscriber = { workspace = true, features = ["env-filter", "json"] }
-tracing-tree = { workspace = true }
 utoipa = { workspace = true, optional = true }
+opentelemetry = { workspace = true, features = ["rt-tokio"], optional = true }
+
 
 [build-dependencies]
 vergen = { workspace = true, features = ["build", "git", "gitcl", "rustc", "cargo"] }
@@ -41,17 +35,13 @@ default = []
 openapi = ["utoipa"]
 output_format = ["serde_json", "dep:clap"]
 bin_info_schema = ["schemars"]
-tokio-console = ["otel"]
-otel = [
-    "chrono",
+basic_tracing = ["dep:tracing", "tracing-subscriber"]
+tracing = [
+    "basic_tracing",
+    "tracing-tree",
+    "opentelemetry-jaeger",
     "tracing-opentelemetry",
     "opentelemetry",
-    "opentelemetry-otlp",
-    "opentelemetry-semantic-conventions",
-    "opentelemetry-stdout",
-    "opentelemetry_sdk",
-    "serde_json",
-    "rand",
 ]
 clap = ["dep:clap", "dep:clap_complete", "dep:clap_complete_fig"]
 models = []

common/bin-common/src/lib.rs

@@ -4,9 +4,6 @@
 pub mod build_information;
 pub mod logging;
 
-#[cfg(feature = "otel")]
-pub mod opentelemetry;
-
 #[cfg(feature = "clap")]
 pub mod completions;
 

common/bin-common/src/logging/error.rs

@@ -1,20 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-#[cfg(feature = "otel")]
-use opentelemetry_otlp::ExporterBuildError;
-
-#[derive(thiserror::Error, Debug)]
-pub enum TracingError {
-    #[error("tracing logger already initialised")]
-    TracingLoggerAlreadyInitialised,
-
-    #[error("Logging error: {0}")]
-    TracingTryInitError(tracing_subscriber::util::TryInitError),
-
-    #[cfg(feature = "otel")]
-    #[error("{0}")]
-    TracingExporterBuildError(#[from] ExporterBuildError),
-
-    #[error("{0}")]
-    TracingFilterParseError(#[from] tracing_subscriber::filter::ParseError),
-}

common/bin-common/src/logging/mod.rs

@@ -1,12 +1,19 @@
 // Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-pub mod error;
-
-use error::TracingError;
 use serde::{Deserialize, Serialize};
 use std::io::IsTerminal;
-use tracing_subscriber::{filter::Directive, layer::SubscriberExt, util::SubscriberInitExt};
+
+#[cfg(feature = "tracing")]
+pub use opentelemetry;
+#[cfg(feature = "tracing")]
+pub use opentelemetry_jaeger;
+#[cfg(feature = "tracing")]
+pub use tracing_opentelemetry;
+#[cfg(feature = "tracing")]
+pub use tracing_subscriber;
+#[cfg(feature = "tracing")]
+pub use tracing_tree;
 
 #[derive(Debug, Default, Copy, Clone, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(deny_unknown_fields)]
@@ -15,6 +22,7 @@ pub struct LoggingSettings {
 }
 
 // don't call init so that we could attach additional layers
+#[cfg(feature = "basic_tracing")]
 pub fn build_tracing_logger() -> impl tracing_subscriber::layer::SubscriberExt {
     use tracing_subscriber::prelude::*;
 
@@ -23,6 +31,7 @@ pub fn build_tracing_logger() -> impl tracing_subscriber::layer::SubscriberExt {
         .with(default_tracing_env_filter())
 }
 
+#[cfg(feature = "basic_tracing")]
 pub fn default_tracing_env_filter() -> tracing_subscriber::filter::EnvFilter {
     if ::std::env::var("RUST_LOG").is_ok() {
         tracing_subscriber::filter::EnvFilter::from_default_env()
@@ -34,6 +43,7 @@ pub fn default_tracing_env_filter() -> tracing_subscriber::filter::EnvFilter {
     }
 }
 
+#[cfg(feature = "basic_tracing")]
 pub fn default_tracing_fmt_layer<S, W>(
     writer: W,
 ) -> impl tracing_subscriber::Layer<S> + Sync + Send + 'static
@@ -53,47 +63,45 @@ where
         .with_target(false)
 }
 
-/// Creates a tracing filter that sets more granular log levels for specific crates.
-/// This allows for finer control over logging verbosity.
-pub(crate) fn granual_filtered_env() -> Result<tracing_subscriber::filter::EnvFilter, TracingError>
-{
-    fn directive_checked(directive: impl Into<String>) -> Result<Directive, TracingError> {
-        directive.into().parse().map_err(From::from)
-    }
-
-    let mut filter = default_tracing_env_filter();
-
-    // these crates are more granularly filtered
-    let filter_crates = ["defguard_wireguard_rs"];
-    for crate_name in filter_crates {
-        filter = filter.add_directive(directive_checked(format!("{crate_name}=warn"))?);
-    }
-    Ok(filter)
-}
-
-pub fn setup_no_otel_logger() -> Result<(), TracingError> {
-    // Only set up if not already initialized
-    if tracing::dispatcher::has_been_set() {
-        // It shouldn't be - this is really checking that it is torn down between async command executions
-        return Err(TracingError::TracingLoggerAlreadyInitialised);
-    }
-
-    let registry = tracing_subscriber::registry()
-        .with(default_tracing_fmt_layer(std::io::stderr))
-        .with(granual_filtered_env()?);
-
-    registry
-        .try_init()
-        .map_err(|e| TracingError::TracingTryInitError(e))?;
-
-    Ok(())
+#[cfg(feature = "basic_tracing")]
+pub fn setup_tracing_logger() {
+    use tracing_subscriber::util::SubscriberInitExt;
+    build_tracing_logger().init()
 }
 
 // TODO: This has to be a macro, running it as a function does not work for the file_appender for some reason
+#[cfg(feature = "tracing")]
 #[macro_export]
 macro_rules! setup_tracing {
     ($service_name: expr) => {
-        setup_no_otel_logger()
+        use nym_bin_common::logging::tracing_subscriber::layer::SubscriberExt;
+        use nym_bin_common::logging::tracing_subscriber::util::SubscriberInitExt;
+
+        let registry = nym_bin_common::logging::tracing_subscriber::Registry::default()
+            .with(nym_bin_common::logging::tracing_subscriber::EnvFilter::from_default_env())
+            .with(
+                nym_bin_common::logging::tracing_tree::HierarchicalLayer::new(4)
+                    .with_targets(true)
+                    .with_bracketed_fields(true),
+            );
+
+        let tracer = nym_bin_common::logging::opentelemetry_jaeger::new_collector_pipeline()
+            .with_endpoint("http://44.199.230.10:14268/api/traces")
+            .with_service_name($service_name)
+            .with_isahc()
+            .with_trace_config(
+                nym_bin_common::logging::opentelemetry::sdk::trace::config().with_sampler(
+                    nym_bin_common::logging::opentelemetry::sdk::trace::Sampler::TraceIdRatioBased(
+                        0.1,
+                    ),
+                ),
+            )
+            .install_batch(nym_bin_common::logging::opentelemetry::runtime::Tokio)
+            .expect("Could not init tracer");
+
+        let telemetry = nym_bin_common::logging::tracing_opentelemetry::layer().with_tracer(tracer);
+
+        registry.with(telemetry).init();
     };
 }
 

common/bin-common/src/opentelemetry/compact_id_generator.rs

@@ -1,47 +0,0 @@
-use opentelemetry_sdk::trace::IdGenerator;
-use opentelemetry::trace::{TraceId, SpanId};
-use rand::RngCore;
-
-#[derive(Clone, Debug)]
-pub struct Compact13BytesIdGenerator;
-
-impl IdGenerator for Compact13BytesIdGenerator {
-    fn new_trace_id(&self) -> TraceId {
-        let mut rng = rand::thread_rng();
-        let mut bytes = [0u8; 16];
-
-        // Fill the first 13 bytes with random data
-        rng.fill_bytes(&mut bytes[0..12]);
-        // Set the last 4 bytes to zero
-        bytes[12] = 0;
-        bytes[13] = 0;
-        bytes[14] = 0;
-        bytes[15] = 0;
-
-        TraceId::from_bytes(bytes)
-    }
-
-    fn new_span_id(&self) -> SpanId {
-        let mut rng = rand::thread_rng();
-        let mut bytes = [0u8; 8];
-        rng.fill_bytes(&mut bytes);
-        
-        SpanId::from_bytes(bytes)
-    }
-}
-
-pub fn compress_trace_id(trace_id: &TraceId) -> [u8; 12] {
-    let bytes = trace_id.to_bytes();
-
-    let mut compressed = [0u8; 12];
-    compressed.copy_from_slice(&bytes[0..12]);
-
-    compressed
-}
-
-pub fn decompress_trace_id(compressed: &[u8; 12]) -> [u8; 16] {
-    let mut bytes = [0u8; 16];
-    bytes[0..12].copy_from_slice(compressed);
-    bytes[12..].copy_from_slice(&[0u8; 4]);
-    bytes
-}

common/bin-common/src/opentelemetry/context.rs

@@ -1,156 +0,0 @@
-use opentelemetry::{Context, TraceFlags};
-use opentelemetry::propagation::{Injector, Extractor, TextMapPropagator};
-use opentelemetry::trace::{SpanContext, TraceContextExt, TraceId};
-use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::IdGenerator};
-use tracing_opentelemetry::OpenTelemetrySpanExt;
-use std::collections::HashMap;
-use std::fmt::Display;
-use tracing::instrument;
-
-/// Make a Carrier for context propagation
-pub struct ContextCarrier {
-    data: HashMap<String, String>,
-}
-
-impl ContextCarrier {
-    pub fn new_empty() -> Self {
-        ContextCarrier {
-            data: HashMap::new(),
-        }
-    }
-
-    pub fn new_with_data(data: HashMap<String, String>) -> Self {
-        if data.is_empty() {
-            return ContextCarrier::new_empty();
-        }
-        
-        ContextCarrier { data }
-    }
-
-    pub fn new_with_current_context(context: Context) -> Self {
-        let mut carrier = ContextCarrier::new_empty();
-        let propagator = TraceContextPropagator::new();
-        propagator.inject_context(&context, &mut carrier);
-        carrier
-    }
-
-    pub fn iter(&self) -> impl Iterator<Item = (&String, &String)> {
-        self.data.iter()
-    }
-
-    pub fn from_map(data: HashMap<String, String>) -> Self {
-        ContextCarrier { data }
-    }
-
-    pub fn into_map(self) -> HashMap<String, String> {
-        self.data
-    }
-
-    pub fn extract_trace_id(&self) -> Option<TraceId> {
-        self.get("traceparent").and_then(|tp| {
-            let parts: Vec<&str> = tp.split('-').collect();
-            if parts.len() == 4 {
-                TraceId::from_hex(parts[1]).ok()
-            } else {
-                None
-            }
-        })
-    }
-
-    pub fn extract_trace_id_into_bytes(&self) -> Option<[u8; 16]> {
-        self.extract_trace_id().map(|id| id.to_bytes())
-    }
-
-    pub fn extract_traceparent(&self) -> Option<String> {
-        self.get("traceparent").map(|s| s.to_string())
-    }
-}
-
-impl Injector for ContextCarrier {
-    fn set(&mut self, key: &str, value: String) {
-        self.data.insert(key.to_string(), value);
-    }
-}
-
-impl Extractor for ContextCarrier {
-    fn get(&self, key: &str) -> Option<&str> {
-        self.data.get(key).map(|s| s.as_str())
-    }
-
-    fn keys(&self) -> Vec<&str> {
-        self.data.keys().map(|k| k.as_str()).collect()
-    }
-}
-
-impl Display for ContextCarrier {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{:?}", self.data)
-    }
-}
-
-pub struct ManualContextPropagator {
-    pub root_span: tracing::Span,
-    pub trace_id: TraceId,
-}
-
-impl ManualContextPropagator {
-    #[instrument(skip_all, level = "debug")]
-    pub fn new(name: &str, context: HashMap<String, String>) -> Self {
-        let carrier = ContextCarrier::new_with_data(context);
-        let trace_id = match carrier.extract_trace_id() {
-            Some(id) => id,
-            None => Context::current().span().span_context().trace_id(),
-        };
-
-        let root_span_builder = new_span_context_with_id(trace_id.clone());
-
-        let root_span = tracing::info_span!("trace_root", name = %name, trace_id = %trace_id);
-        root_span.set_parent(root_span_builder);
-
-        ManualContextPropagator {
-            root_span,
-            trace_id,
-        }
-    }
-
-    #[instrument(skip_all, level = "debug")]
-    pub fn new_from_tid(name: &str, trace_id: TraceId) -> Self {
-        let root_span_builder = new_span_context_with_id(trace_id.clone());
-
-        let root_span = tracing::info_span!("trace_root", name = %name, trace_id = %trace_id);
-        root_span.set_parent(root_span_builder);
-
-        ManualContextPropagator {
-            root_span,
-            trace_id,
-        }
-    }
-
-    pub fn root_span(&self) -> &tracing::Span {
-        &self.root_span
-    }
-
-}
-
-#[instrument(skip_all, level = "debug")]
-pub fn new_span_context_with_id(trace_id: TraceId) -> Context {
-    let id_gen = opentelemetry_sdk::trace::RandomIdGenerator::default();
-    let span_id = id_gen.new_span_id();
-    let span_context = SpanContext::new(
-        trace_id,
-        span_id,
-        TraceFlags::SAMPLED,
-        true,
-        Default::default(),
-    );
-
-    Context::current().with_remote_span_context(span_context)
-}
-
-#[instrument(skip_all, level = "debug")]
-pub fn extract_trace_id_from_tracing_cx() -> TraceId {
-    let cx = tracing::Span::current().context();
-    let binding = cx.span();
-    let trace_id = binding.span_context().trace_id();
-    trace_id
-}

common/bin-common/src/opentelemetry/mod.rs

@@ -1,308 +0,0 @@
-pub mod context;
-pub mod compact_id_generator;
-mod trace_id_format;
-
-use tracing::{info, Level};
-use tracing_subscriber::filter::Directive;
-use tracing_subscriber::layer::SubscriberExt;
-use tracing_subscriber::util::SubscriberInitExt;
-use tracing_subscriber::fmt;
-
-use crate::logging::default_tracing_env_filter;
-use crate::logging::error::TracingError;
-use crate::opentelemetry::compact_id_generator::Compact13BytesIdGenerator;
-use opentelemetry::trace::TracerProvider;
-use opentelemetry::{global, KeyValue};
-use opentelemetry_otlp::tonic_types::metadata::MetadataMap;
-use opentelemetry_otlp::tonic_types::transport::ClientTlsConfig;
-use opentelemetry_otlp::{WithExportConfig, WithTonicConfig};
-use opentelemetry_sdk::metrics::{MeterProviderBuilder, PeriodicReader, SdkMeterProvider};
-use opentelemetry_sdk::trace::SdkTracerProvider;
-use opentelemetry_sdk::{trace::Sampler, Resource};
-use opentelemetry_semantic_conventions::resource::{DEPLOYMENT_ENVIRONMENT_NAME, SERVICE_VERSION};
-use opentelemetry_semantic_conventions::SCHEMA_URL;
-use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
-use tracing_subscriber::fmt::format::FmtSpan;
-
-pub struct TracerProviderGuard(Option<SdkTracerProvider>);
-
-impl Drop for TracerProviderGuard {
-    fn drop(&mut self) {
-        if let Some(tracer_provider) = self.0.take() {
-            // Ensure all spans are flushed before exit
-            if let Err(e) = tracer_provider.shutdown() {
-                eprintln!("Error shutting down tracer provider: {:?}", e);
-            }
-        }
-    }
-}
-
-pub(crate) fn granual_filtered_env() -> Result<tracing_subscriber::filter::EnvFilter, TracingError>
-{
-    fn directive_checked(directive: impl Into<String>) -> Result<Directive, TracingError> {
-        directive.into().parse().map_err(From::from)
-    }
-
-    let mut filter = default_tracing_env_filter();
-
-    // these crates are more granularly filtered
-    let filter_crates = ["defguard_wireguard_rs"];
-    for crate_name in filter_crates {
-        filter = filter.add_directive(directive_checked(format!("{crate_name}=warn"))?);
-    }
-    Ok(filter)
-}
-
-pub fn setup_tracing_logger(service_name: String) -> Result<TracerProviderGuard, TracingError> {
-    if tracing::dispatcher::has_been_set() {
-        // It shouldn't be - this is really checking that it is torn down between async command executions
-        return Err(TracingError::TracingLoggerAlreadyInitialised);
-    }
-
-    // define ingestion points
-    let endpoint = std::env::var("SIGNOZ_ENDPOINT").expect("SIGNOZ_ENDPOINT not set");
-    let key = std::env::var("SIGNOZ_INGESTION_KEY").expect("SIGNOZ_INGESTION_KEY not set");
-    let mut metadata = MetadataMap::new();
-    metadata.insert(
-        "signoz-ingestion-key",
-        key.parse().expect("Could not parse signoz ingestion key"),
-    );
-
-    // Build resources
-    let resource = build_resource(&service_name);
-
-    // Initialize tracer and meter providers
-    let tracer_provider = init_tracer_provider(&endpoint, metadata.clone(), resource.clone())?;
-    let meter_provider = init_meter_provider(&endpoint, metadata.clone(), resource.clone())?;
-
-    // Bridge tracing and opentelemetry
-    let tracer = tracer_provider.tracer("otel-subscriber");
-    let fmt_layer = fmt::layer()
-        .json()
-        .with_writer(std::io::stderr)
-        .with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
-        .with_span_list(false)
-        .with_current_span(true)
-        .event_format(trace_id_format::TraceIdFormat);
-
-    let registry = tracing_subscriber::registry()
-        .with(fmt_layer)
-        .with(granual_filtered_env()?)
-        .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO))
-        .with(MetricsLayer::new(meter_provider.clone()))
-        .with(OpenTelemetryLayer::new(tracer));
-
-    registry.try_init().map_err(TracingError::TracingTryInitError)?;
-
-    global::set_tracer_provider(tracer_provider.clone());
-    global::set_meter_provider(meter_provider.clone());
-
-    info!("Tracing initialized with service name: {}", service_name);
-
-    Ok(TracerProviderGuard(Some(tracer_provider)))
-}
-
-fn build_resource(service_name: &str) -> Resource {
-    Resource::builder()
-        .with_service_name(service_name.to_string())
-        .with_schema_url(
-            [
-                KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
-                KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, "develop"),
-            ],
-            SCHEMA_URL,
-        )
-        .build()
-}
-
-fn init_tracer_provider(
-    endpoint: &str,
-    metadata: MetadataMap,
-    resource: Resource,
-) -> Result<SdkTracerProvider, TracingError> {
-    let mut exporter_builder = opentelemetry_otlp::SpanExporter::builder()
-        .with_tonic()
-        .with_metadata(metadata)
-        .with_endpoint(endpoint);
-
-    if endpoint.starts_with("https://") {
-        exporter_builder =
-            exporter_builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots());
-    }
-
-    let exporter = exporter_builder.build()?;
-
-    let tracer = SdkTracerProvider::builder()
-        .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
-            1.0,
-        ))))
-        .with_id_generator(Compact13BytesIdGenerator)
-        .with_resource(resource)
-        .with_batch_exporter(exporter)
-        .build();
-
-    global::set_tracer_provider(tracer.clone());
-    Ok(tracer)
-}
-
-fn init_meter_provider(
-    endpoint: &str,
-    metadata: MetadataMap,
-    resource: Resource,
-) -> Result<SdkMeterProvider, TracingError> {
-    let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder()
-        .with_tonic()
-        .with_metadata(metadata)
-        .with_endpoint(endpoint)
-        .with_temporality(opentelemetry_sdk::metrics::Temporality::default());
-
-    if endpoint.starts_with("https://") {
-        exporter_builder = exporter_builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots());
-    }
-
-    let exporter = exporter_builder.build()?;
-
-    let reader = PeriodicReader::builder(exporter)
-        .with_interval(std::time::Duration::from_secs(30))
-        .build();
-
-    let stdout_reader =
-        PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()).build();
-
-    let meter_provider = MeterProviderBuilder::default()
-        .with_resource(resource)
-        .with_reader(reader)
-        .with_reader(stdout_reader)
-        .build();
-
-    global::set_meter_provider(meter_provider.clone());
-
-    Ok(meter_provider)
-}
-
-// pub fn setup_tracing_logger(service_name: String) -> Result<(), TracingError> {
-//     if tracing::dispatcher::has_been_set() {
-//         // It shouldn't be - this is really checking that it is torn down between async command executions
-//         return Err(TracingError::TracingLoggerAlreadyInitialised);
-//     }
-
-//     let key =
-//         std::env::var("SIGNOZ_INGESTION_KEY".to_string()).expect("SIGNOZ_INGESTION_KEY not set");
-//     let mut metadata = MetadataMap::new();
-//     metadata.insert(
-//         "signoz-ingestion-key",
-//         key.parse().expect("Could not parse signoz ingestion key"),
-//     );
-
-//     let tracer_provider = init_tracer_provider(metadata.clone(), service_name.clone())?;
-//     let meter_provider = init_meter_provider(metadata.clone(), service_name.clone())?;
-//     let tracer = tracer_provider.tracer("tracing-otel-subscriber");
-//     let fmt_layer = fmt::layer()
-//         .json()
-//         .with_writer(std::io::stderr)
-//         .with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
-//         .with_span_list(false)
-//         .with_current_span(true)
-//         .event_format(trace_id_format::TraceIdFormat);
-
-//     cfg_if::cfg_if! {if #[cfg(feature = "tokio-console")] {
-//         // instrument tokio console subscriber needs RUSTFLAGS="--cfg tokio_unstable" at build time
-//         let console_layer = console_subscriber::spawn();
-
-//         tracing_subscriber::registry()
-//             .with(console_layer)
-//             .with(fmt_layer)
-//             .with(granual_filtered_env()?)
-//             .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO))
-//             .with(MetricsLayer::new(meter_provider))
-//             .with(OpenTelemetryLayer::new(tracer))
-//             .try_init()
-//             .map_err(|e| TracingError::TracingTryInitError(e))?;
-//     } else {
-//         tracing_subscriber::registry()
-//             .with(fmt_layer)
-//             .with(granual_filtered_env()?)
-//             .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO))
-//             .with(MetricsLayer::new(meter_provider))
-//             .with(OpenTelemetryLayer::new(tracer))
-//             .try_init()
-//             .map_err(|e| TracingError::TracingTryInitError(e))?;
-//     }}
-
-//     Ok(())
-// }
-
-// fn resource(service_name: String) -> Resource {
-//     Resource::builder()
-//         .with_service_name(service_name)
-//         .with_schema_url(
-//             [
-//                 KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
-//                 KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, "develop"),
-//             ],
-//             SCHEMA_URL,
-//         )
-//         .build()
-// }
-
-// fn init_tracer_provider(metadata: MetadataMap, service_name: String) -> Result<SdkTracerProvider, TracingError> {
-//     let endpoint = std::env::var("SIGNOZ_ENDPOINT".to_string()).expect("SIGNOZ_ENDPOINT not set");
-//     info!("SIGNOZ_ENDPOINT = {}", endpoint);
-
-//     let mut exporter_builder = opentelemetry_otlp::SpanExporter::builder()
-//         .with_tonic()
-//         .with_metadata(metadata)
-//         .with_endpoint(&endpoint);
-
-//     if endpoint.starts_with("https://") {
-//         exporter_builder =
-//             exporter_builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots());
-//     }
-
-//     let exporter = exporter_builder.build()?;
-
-//     let tracer = SdkTracerProvider::builder()
-//         .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
-//             1.0,
-//         ))))
-//         .with_id_generator(Compact13BytesIdGenerator)
-//         .with_resource(resource(service_name))
-//         .with_batch_exporter(exporter)
-//         .build();
-
-//     global::set_tracer_provider(tracer.clone());
-//     Ok(tracer)
-// }
-
-// fn init_meter_provider(metadata: MetadataMap, service_name: String) -> Result<SdkMeterProvider, TracingError> {
-//     let endpoint = std::env::var("SIGNOZ_ENDPOINT".to_string()).expect("SIGNOZ_ENDPOINT not set");
-
-//     let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder()
-//         .with_tonic()
-//         .with_metadata(metadata)
-//         .with_endpoint(&endpoint)
-//         .with_temporality(opentelemetry_sdk::metrics::Temporality::default());
-
-//     if endpoint.starts_with("https://") {
-//         exporter_builder = exporter_builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots());
-//     }
-
-//     let exporter = exporter_builder.build()?;
-
-//     let reader = PeriodicReader::builder(exporter)
-//         .with_interval(std::time::Duration::from_secs(30))
-//         .build();
-
-//     let stdout_reader =
-//         PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()).build();
-
-//     let meter_provider = MeterProviderBuilder::default()
-//         .with_resource(resource(service_name))
-//         .with_reader(reader)
-//         .with_reader(stdout_reader)
-//         .build();
-
-//     global::set_meter_provider(meter_provider.clone());
-
-//     Ok(meter_provider)
-// }

common/bin-common/src/opentelemetry/trace_id_format.rs

@@ -1,88 +0,0 @@
-use chrono::Utc;
-use opentelemetry::trace::TraceContextExt;
-use opentelemetry::{SpanId, TraceId};
-use serde::ser::{SerializeMap, Serializer as _};
-use std::io;
-use tracing::{Event, Subscriber};
-use tracing_opentelemetry::OpenTelemetrySpanExt;
-use tracing_serde::fields::AsMap;
-use tracing_serde::AsSerde;
-use tracing_subscriber::fmt::format::Writer;
-use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields};
-use tracing_subscriber::registry::LookupSpan;
-
-pub struct WriteAdaptor<'a> {
-    fmt_write: &'a mut dyn std::fmt::Write,
-}
-
-impl<'a> WriteAdaptor<'a> {
-    pub fn new(fmt_write: &'a mut dyn std::fmt::Write) -> Self {
-        Self { fmt_write }
-    }
-}
-
-impl<'a> io::Write for WriteAdaptor<'a> {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        let s =
-            std::str::from_utf8(buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
-
-        self.fmt_write
-            .write_str(s)
-            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
-
-        Ok(s.as_bytes().len())
-    }
-
-    fn flush(&mut self) -> io::Result<()> {
-        Ok(())
-    }
-}
-
-pub struct TraceIdFormat;
-
-impl<S, N> FormatEvent<S, N> for TraceIdFormat
-where
-    S: Subscriber + for<'lookup> LookupSpan<'lookup>,
-    N: for<'writer> FormatFields<'writer> + 'static,
-{
-    fn format_event(
-        &self,
-        _ctx: &FmtContext<'_, S, N>,
-        mut writer: Writer<'_>,
-        event: &Event<'_>,
-    ) -> std::fmt::Result
-    where
-        S: Subscriber + for<'a> LookupSpan<'a>,
-    {
-        let meta = event.metadata();
-
-        let mut visit = || {
-            let mut serializer = serde_json::Serializer::new(WriteAdaptor::new(&mut writer));
-            let mut serializer = serializer.serialize_map(None)?;
-            serializer.serialize_entry("timestamp", &Utc::now().to_rfc3339())?;
-            serializer.serialize_entry("level", &meta.level().as_serde())?;
-            serializer.serialize_entry("fields", &event.field_map())?;
-            serializer.serialize_entry("target", meta.target())?;
-
-            let current_span = tracing::Span::current();
-            let context = current_span.context();
-            let span_ref = context.span();
-            let span_context = span_ref.span_context();
-
-            let trace_id = span_context.trace_id();
-            if trace_id != TraceId::INVALID {
-                serializer.serialize_entry("trace_id", &trace_id.to_string())?;
-
-                let span_id = span_context.span_id();
-                if span_id != SpanId::INVALID {
-                    serializer.serialize_entry("span_id", &span_id.to_string())?;
-                }
-            }
-
-            serializer.end()
-        };
-
-        visit().map_err(|_| std::fmt::Error)?;
-        writeln!(writer)
-    }
-}

common/client-core/Cargo.toml

@@ -13,7 +13,6 @@ async-trait = { workspace = true }
 base64 = { workspace = true }
 bs58 = { workspace = true }
 clap = { workspace = true, optional = true }
-cfg-if = { workspace = true }
 comfy-table = { workspace = true, optional = true }
 futures = { workspace = true }
 humantime = { workspace = true }
@@ -53,7 +52,6 @@ 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 }
@@ -123,9 +121,5 @@ cli = ["clap", "comfy-table"]
 fs-credentials-storage = ["nym-credential-storage/persistent-storage"]
 fs-surb-storage = ["nym-client-core-surb-storage/fs-surb-storage"]
 fs-gateways-storage = ["nym-client-core-gateways-storage/fs-gateways-storage"]
-otel = ["nym-sphinx/otel"]
 wasm = ["nym-gateway-client/wasm"]
 metrics-server = []
-
-[lints]
-workspace = true

common/client-core/gateways-storage/Cargo.toml

@@ -3,7 +3,6 @@ name = "nym-client-core-gateways-storage"
 version = "0.1.0"
 edition = "2021"
 license.workspace = true
-rust-version.workspace = true
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -27,7 +26,6 @@ features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "time"]
 optional = true
 
 [build-dependencies]
-anyhow = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 sqlx = { workspace = true, features = [
     "runtime-tokio-rustls",

common/client-core/gateways-storage/build.rs

@@ -2,30 +2,23 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #[tokio::main]
-async fn main() -> anyhow::Result<()> {
+async fn main() {
     #[cfg(feature = "fs-gateways-storage")]
     {
-        use anyhow::Context;
         use sqlx::{Connection, SqliteConnection};
         use std::env;
 
-        let out_dir = env::var("OUT_DIR")?;
+        let out_dir = env::var("OUT_DIR").unwrap();
         let database_path = format!("{out_dir}/gateways-storage-example.sqlite");
 
-        // remove the db file if it already existed from previous build
-        // in case it was from a different branch
-        if std::fs::exists(&database_path)? {
-            std::fs::remove_file(&database_path)?;
-        }
-
         let mut conn = SqliteConnection::connect(&format!("sqlite://{database_path}?mode=rwc"))
             .await
-            .context("Failed to create SQLx database connection")?;
+            .expect("Failed to create SQLx database connection");
 
         sqlx::migrate!("./fs_gateways_migrations")
             .run(&mut conn)
             .await
-            .context("Failed to perform SQLx migrations")?;
+            .expect("Failed to perform SQLx migrations");
 
         #[cfg(target_family = "unix")]
         println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path);
@@ -35,6 +28,4 @@ async fn main() -> anyhow::Result<()> {
         // not a valid windows path... but hey, it works...
         println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
     }
-
-    Ok(())
 }

common/client-core/src/cli_helpers/client_import_coin_index_signatures.rs

@@ -58,7 +58,6 @@ where
         Some(data) => data,
         None => {
             // SAFETY: one of those arguments must have been set
-            #[allow(clippy::unwrap_used)]
             fs::read(common_args.signatures_path.unwrap())?
         }
     };

common/client-core/src/cli_helpers/client_import_credential.rs

@@ -64,7 +64,6 @@ where
         Some(data) => data,
         None => {
             // SAFETY: one of those arguments must have been set
-            #[allow(clippy::unwrap_used)]
             fs::read(common_args.credential_path.unwrap())?
         }
     };

common/client-core/src/cli_helpers/client_import_expiration_date_signatures.rs

@@ -58,7 +58,6 @@ where
         Some(data) => data,
         None => {
             // SAFETY: one of those arguments must have been set
-            #[allow(clippy::unwrap_used)]
             fs::read(common_args.signatures_path.unwrap())?
         }
     };

common/client-core/src/cli_helpers/client_import_master_verification_key.rs

@@ -58,7 +58,6 @@ where
         Some(data) => data,
         None => {
             // SAFETY: one of those arguments must have been set
-            #[allow(clippy::unwrap_used)]
             fs::read(common_args.key_path.unwrap())?
         }
     };

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,15 +48,16 @@ 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::ShutdownTracker;
+use nym_task::{TaskClient, TaskHandle};
 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, UserAgent};
+use nym_validator_client::{nyxd::contract_traits::DkgQueryClient, NymApiClient, UserAgent};
 use rand::prelude::SliceRandom;
 use rand::rngs::OsRng;
 use rand::thread_rng;
@@ -94,12 +95,12 @@ impl ClientInput {
     }
 }
 
+#[derive(Clone)]
 pub struct ClientOutput {
     pub received_buffer_request_sender: ReceivedBufferRequestSender,
 }
 
 impl ClientOutput {
-    #[instrument(name = "ClientOutput::register_receiver", skip_all)]
     pub fn register_receiver(
         &mut self,
     ) -> Result<mpsc::UnboundedReceiver<Vec<ReconstructedMessage>>, ClientCoreError> {
@@ -134,11 +135,9 @@ pub enum ClientInputStatus {
 }
 
 impl ClientInputStatus {
-    #[allow(clippy::panic)]
     pub fn register_producer(&mut self) -> ClientInput {
         match std::mem::replace(self, ClientInputStatus::Connected) {
             ClientInputStatus::AwaitingProducer { client_input } => client_input,
-            // critical failure implying misuse of software
             ClientInputStatus::Connected => panic!("producer was already registered before"),
         }
     }
@@ -150,11 +149,9 @@ pub enum ClientOutputStatus {
 }
 
 impl ClientOutputStatus {
-    #[allow(clippy::panic)]
     pub fn register_consumer(&mut self) -> ClientOutput {
         match std::mem::replace(self, ClientOutputStatus::Connected) {
             ClientOutputStatus::AwaitingConsumer { client_output } => client_output,
-            // critical failure implying misuse of software
             ClientOutputStatus::Connected => panic!("consumer was already registered before"),
         }
     }
@@ -194,7 +191,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<ShutdownTracker>,
+    shutdown: Option<TaskClient>,
     user_agent: Option<UserAgent>,
 
     setup_method: GatewaySetup,
@@ -280,7 +277,7 @@ where
     }
 
     #[must_use]
-    pub fn with_shutdown(mut self, shutdown: ShutdownTracker) -> Self {
+    pub fn with_shutdown(mut self, shutdown: TaskClient) -> Self {
         self.shutdown = Some(shutdown);
         self
     }
@@ -324,11 +321,11 @@ where
         topology_accessor: TopologyAccessor,
         mix_tx: BatchMixMessageSender,
         stats_tx: ClientStatsSender,
-        shutdown_tracker: &ShutdownTracker,
+        task_client: TaskClient,
     ) {
         info!("Starting loop cover traffic stream...");
 
-        let mut stream = LoopCoverTrafficStream::new(
+        let stream = LoopCoverTrafficStream::new(
             ack_key,
             debug_config.acknowledgements.average_ack_delay,
             mix_tx,
@@ -337,9 +334,10 @@ where
             debug_config.traffic,
             debug_config.cover_traffic,
             stats_tx,
+            task_client,
         );
-        shutdown_tracker
-            .try_spawn_named_with_shutdown(async move { stream.run().await }, "CoverTrafficStream");
+
+        stream.start();
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -355,12 +353,13 @@ 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...");
 
-        let real_messages_controller = RealMessagesController::new(
+        RealMessagesController::new(
             controller_config,
             key_rotation_config,
             ack_receiver,
@@ -373,101 +372,37 @@ where
             lane_queue_lengths,
             client_connection_rx,
             stats_tx,
-            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);
+            task_client,
+        )
+        .start(packet_type);
     }
 
     // buffer controlling all messages fetched from provider
     // required so that other components would be able to use them (say the websocket)
-    #[instrument(skip_all)]
     fn start_received_messages_buffer_controller(
         local_encryption_keypair: Arc<x25519::KeyPair>,
         query_receiver: ReceivedBufferRequestReceiver,
         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>::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",
-        );
+        let controller: ReceivedMessagesBufferController<SphinxMessageReceiver> =
+            ReceivedMessagesBufferController::new(
+                local_encryption_keypair,
+                query_receiver,
+                mixnet_receiver,
+                reply_key_storage,
+                reply_controller_sender,
+                metrics_reporter,
+                shutdown,
+            );
+        controller.start()
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[instrument(skip_all)]
     async fn start_gateway_client(
         config: &Config,
         initialisation_result: InitialisationResult,
@@ -476,7 +411,7 @@ where
         packet_router: PacketRouter,
         stats_reporter: ClientStatsSender,
         #[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
-        shutdown_tracker: &ShutdownTracker,
+        shutdown: TaskClient,
     ) -> Result<GatewayClient<C, S::CredentialStore>, ClientCoreError>
     where
         <S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
@@ -495,7 +430,7 @@ where
                     packet_router,
                     bandwidth_controller,
                     stats_reporter,
-                    shutdown_tracker.clone_shutdown_token(),
+                    shutdown,
                 )
             } else {
                 let cfg = GatewayConfig::new(
@@ -520,7 +455,7 @@ where
                     stats_reporter,
                     #[cfg(unix)]
                     connection_fd_callback,
-                    shutdown_tracker.clone_shutdown_token(),
+                    shutdown,
                 )
             };
 
@@ -574,7 +509,6 @@ where
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[instrument(skip_all)]
     async fn setup_gateway_transceiver(
         custom_gateway_transceiver: Option<Box<dyn GatewayTransceiver + Send>>,
         config: &Config,
@@ -584,7 +518,7 @@ where
         packet_router: PacketRouter,
         stats_reporter: ClientStatsSender,
         #[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
-        shutdown_tracker: &ShutdownTracker,
+        mut shutdown: TaskClient,
     ) -> Result<Box<dyn GatewayTransceiver + Send>, ClientCoreError>
     where
         <S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
@@ -601,6 +535,7 @@ 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)
             };
@@ -616,7 +551,7 @@ where
             stats_reporter,
             #[cfg(unix)]
             connection_fd_callback,
-            shutdown_tracker,
+            shutdown,
         )
         .await?;
 
@@ -627,7 +562,7 @@ where
         custom_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
         config_topology: config::Topology,
         nym_api_urls: Vec<Url>,
-        nym_api_client: nym_http_api_client::Client,
+        nym_api_client: NymApiClient,
     ) -> Box<dyn TopologyProvider + Send + Sync> {
         // if no custom provider was ... provided ..., create one using nym-api
         custom_provider.unwrap_or_else(|| {
@@ -647,20 +582,22 @@ where
         topology_accessor: TopologyAccessor,
         local_gateway: NodeIdentity,
         wait_for_gateway: bool,
-        shutdown_tracker: &ShutdownTracker,
+        mut task_client: TaskClient,
     ) -> 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 refresher is not going to be started");
+            info!("The background topology refesher is not going to be started");
+            task_client.disarm();
         }
 
         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
@@ -705,10 +642,7 @@ 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...");
-            shutdown_tracker.try_spawn_named_with_shutdown(
-                async move { topology_refresher.run().await },
-                "TopologyRefresher",
-            );
+            topology_refresher.start();
         }
 
         Ok(())
@@ -719,7 +653,7 @@ where
         user_agent: Option<UserAgent>,
         client_stats_id: String,
         input_sender: Sender<InputMessage>,
-        shutdown_tracker: &ShutdownTracker,
+        task_client: TaskClient,
     ) -> ClientStatsSender {
         info!("Starting statistics control...");
         StatisticsControl::create_and_start(
@@ -729,23 +663,18 @@ where
                 .unwrap_or("unknown".to_string()),
             client_stats_id,
             input_sender.clone(),
-            shutdown_tracker,
+            task_client,
         )
     }
-    #[instrument(skip_all)]
+
     fn start_mix_traffic_controller(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
-        shutdown_tracker: &ShutdownTracker,
+        shutdown: TaskClient,
     ) -> (BatchMixMessageSender, ClientRequestSender) {
         info!("Starting mix traffic controller...");
-        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",
-        );
-
+        let (mix_traffic_controller, mix_tx, client_tx) =
+            MixTrafficController::new(gateway_transceiver, shutdown);
+        mix_traffic_controller.start();
         (mix_tx, client_tx)
     }
 
@@ -753,7 +682,7 @@ where
     async fn setup_persistent_reply_storage(
         backend: S::ReplyStore,
         key_rotation_config: KeyRotationConfig,
-        shutdown_tracker: &ShutdownTracker,
+        shutdown: TaskClient,
     ) -> Result<CombinedReplyStorage, ClientCoreError>
     where
         <S::ReplyStore as ReplyStorageBackend>::StorageError: Sync + Send,
@@ -778,15 +707,11 @@ where
             })?;
 
         let store_clone = mem_store.clone();
-        let shutdown_token = shutdown_tracker.clone_shutdown_token();
-        shutdown_tracker.try_spawn_named(
-            async move {
-                persistent_storage
-                    .flush_on_shutdown(store_clone, shutdown_token)
-                    .await
-            },
-            "PersistentReplyStorage::flush_on_shutdown",
-        );
+        spawn_future(async move {
+            persistent_storage
+                .flush_on_shutdown(store_clone, shutdown)
+                .await
+        });
 
         Ok(mem_store)
     }
@@ -807,7 +732,7 @@ where
             let mut rng = OsRng;
             let keys = if let Some(derivation_material) = derivation_material {
                 ClientKeys::from_master_key(&mut rng, &derivation_material)
-                    .map_err(|_| ClientCoreError::HkdfDerivationError)?
+                    .map_err(|_| ClientCoreError::HkdfDerivationError {})?
             } else {
                 ClientKeys::generate_new(&mut rng)
             };
@@ -817,32 +742,23 @@ where
         setup_gateway(setup_method, key_store, details_store).await
     }
 
-    fn construct_nym_api_client(
-        config: &Config,
-        user_agent: Option<UserAgent>,
-    ) -> Result<nym_http_api_client::Client, ClientCoreError> {
+    fn construct_nym_api_client(config: &Config, user_agent: Option<UserAgent>) -> NymApiClient {
         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 {
-            builder = builder.with_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_bincode();
-
-        builder.build().map_err(ClientCoreError::from)
     }
 
     async fn determine_key_rotation_state(
-        client: &nym_http_api_client::Client,
+        client: &NymApiClient,
     ) -> Result<KeyRotationConfig, ClientCoreError> {
-        Ok(client.get_key_rotation_info().await?.into())
+        Ok(client.nym_api.get_key_rotation_info().await?.into())
     }
 
-    #[instrument(skip_all)]
     pub async fn start_base(mut self) -> Result<BaseClient, ClientCoreError>
     where
         S::ReplyStore: Send + Sync,
@@ -885,12 +801,12 @@ where
         let shared_topology_accessor =
             TopologyAccessor::new(self.config.debug.topology.ignore_egress_epoch_role);
 
-        // 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()?,
-        };
+        // Shutdown notifier for signalling tasks to stop
+        let shutdown = self
+            .shutdown
+            .map(Into::<TaskHandle>::into)
+            .unwrap_or_default()
+            .name_if_unnamed("BaseNymClient");
 
         // channels responsible for dealing with reply-related fun
         let (reply_controller_sender, reply_controller_receiver) =
@@ -907,7 +823,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(
@@ -922,7 +838,7 @@ where
             self.user_agent.clone(),
             generate_client_stats_id(*self_address.identity()),
             input_sender.clone(),
-            &shutdown_tracker.child_tracker(),
+            shutdown.fork("statistics_control"),
         );
 
         // needs to be started as the first thing to block if required waiting for the gateway
@@ -932,14 +848,14 @@ where
             shared_topology_accessor.clone(),
             self_address.gateway(),
             self.wait_for_gateway,
-            &shutdown_tracker.child_tracker(),
+            shutdown.fork("topology_refresher"),
         )
         .await?;
 
         let gateway_packet_router = PacketRouter::new(
             ack_sender,
             mixnet_messages_sender,
-            shutdown_tracker.clone_shutdown_token(),
+            shutdown.get_handle().named("gateway-packet-router"),
         );
 
         let gateway_transceiver = Self::setup_gateway_transceiver(
@@ -952,7 +868,7 @@ where
             stats_reporter.clone(),
             #[cfg(unix)]
             self.connection_fd_callback,
-            &shutdown_tracker.child_tracker(),
+            shutdown.fork("gateway_transceiver"),
         )
         .await?;
         let gateway_ws_fd = gateway_transceiver.ws_fd();
@@ -960,7 +876,7 @@ where
         let reply_storage = Self::setup_persistent_reply_storage(
             reply_storage_backend,
             key_rotation_config,
-            &shutdown_tracker.child_tracker(),
+            shutdown.fork("persistent_reply_storage"),
         )
         .await?;
 
@@ -970,8 +886,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
@@ -981,7 +897,7 @@ where
 
         let (message_sender, client_request_sender) = Self::start_mix_traffic_controller(
             gateway_transceiver,
-            &shutdown_tracker.child_tracker(),
+            shutdown.fork("mix_traffic_controller"),
         );
 
         // Channels that the websocket listener can use to signal downstream to the real traffic
@@ -1010,8 +926,9 @@ 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
@@ -1027,7 +944,7 @@ where
                 shared_topology_accessor.clone(),
                 message_sender,
                 stats_reporter.clone(),
-                &shutdown_tracker.child_tracker(),
+                shutdown.fork("cover_traffic_stream"),
             );
         }
 
@@ -1055,7 +972,7 @@ where
                 gateway_connection: GatewayConnection { gateway_ws_fd },
             },
             stats_reporter,
-            shutdown_handle: Some(shutdown_tracker), // The primary tracker for this client
+            task_handle: shutdown,
             client_request_sender,
             forget_me: self.config.debug.forget_me,
             remember_me: self.config.debug.remember_me,
@@ -1071,7 +988,7 @@ pub struct BaseClient {
     pub client_state: ClientState,
     pub stats_reporter: ClientStatsSender,
     pub client_request_sender: ClientRequestSender,
-    pub shutdown_handle: Option<ShutdownTracker>,
+    pub task_handle: TaskHandle,
     pub forget_me: ForgetMe,
     pub remember_me: RememberMe,
 }

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

@@ -114,32 +114,41 @@ pub async fn setup_fs_gateways_storage<P: AsRef<Path>>(
         })
 }
 
-pub fn create_bandwidth_controller_with_urls<St: CredentialStorage>(
-    nyxd_url: Url,
-    storage: St,
-) -> Result<BandwidthController<QueryHttpRpcNyxdClient, St>, ClientCoreError> {
-    let client = default_query_dkg_client(nyxd_url)?;
-
-    Ok(BandwidthController::new(storage, client))
-}
-
-pub fn default_query_dkg_client_from_config(
+pub fn create_bandwidth_controller<St: CredentialStorage>(
     config: &Config,
-) -> Result<QueryHttpRpcNyxdClient, ClientCoreError> {
+    storage: St,
+) -> BandwidthController<QueryHttpRpcNyxdClient, St> {
     let nyxd_url = config
         .get_validator_endpoints()
         .pop()
-        .ok_or(ClientCoreError::RpcClientMissingUrl)?;
+        .expect("No nyxd validator endpoint provided");
+
+    create_bandwidth_controller_with_urls(nyxd_url, storage)
+}
+
+pub fn create_bandwidth_controller_with_urls<St: CredentialStorage>(
+    nyxd_url: Url,
+    storage: St,
+) -> BandwidthController<QueryHttpRpcNyxdClient, St> {
+    let client = default_query_dkg_client(nyxd_url);
+
+    BandwidthController::new(storage, client)
+}
+
+pub fn default_query_dkg_client_from_config(config: &Config) -> QueryHttpRpcNyxdClient {
+    let nyxd_url = config
+        .get_validator_endpoints()
+        .pop()
+        .expect("No nyxd validator endpoint provided");
 
     default_query_dkg_client(nyxd_url)
 }
 
-pub fn default_query_dkg_client(nyxd_url: Url) -> Result<QueryHttpRpcNyxdClient, ClientCoreError> {
+pub fn default_query_dkg_client(nyxd_url: Url) -> QueryHttpRpcNyxdClient {
     let details = nym_network_defaults::NymNetworkDetails::new_from_env();
     let client_config = nyxd::Config::try_from_nym_network_details(&details)
-        .map_err(|source| ClientCoreError::InvalidNetworkDetails { source })?;
+        .expect("failed to construct validator client config");
     // overwrite env configuration with config URLs
-
     QueryHttpRpcNyxdClient::connect(client_config, nyxd_url.as_str())
-        .map_err(|source| ClientCoreError::RpcClientCreationFailure { source })
+        .expect("Could not construct query client")
 }

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;
+use crate::{config, spawn_future};
 use futures::task::{Context, Poll};
 use futures::{Future, Stream, StreamExt};
 use nym_sphinx::acknowledgements::AckKey;
@@ -12,6 +12,7 @@ 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;
@@ -68,6 +69,8 @@ where
     packet_type: PacketType,
 
     stats_tx: ClientStatsSender,
+
+    task_client: TaskClient,
 }
 
 impl<R> Stream for LoopCoverTrafficStream<R>
@@ -114,6 +117,7 @@ impl LoopCoverTrafficStream<OsRng> {
         traffic_config: config::Traffic,
         cover_config: config::CoverTraffic,
         stats_tx: ClientStatsSender,
+        task_client: TaskClient,
     ) -> Self {
         let rng = OsRng;
 
@@ -133,6 +137,7 @@ impl LoopCoverTrafficStream<OsRng> {
             use_legacy_sphinx_format: traffic_config.use_legacy_sphinx_format,
             packet_type: traffic_config.packet_type,
             stats_tx,
+            task_client,
         }
     }
 
@@ -230,13 +235,11 @@ 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 async fn run(&mut self) {
+    pub fn start(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 run LoopCoverTrafficStream while config explicitly disabled it.")
+            panic!("attempted to start LoopCoverTrafficStream while config explicitly disabled it.")
         }
 
         // we should set initial delay only when we actually start the stream
@@ -246,11 +249,29 @@ impl LoopCoverTrafficStream<OsRng> {
         );
         self.set_next_delay(sampled);
 
-        while self.next().await.is_some() {
-            self.on_new_message().await;
-        }
+        let mut shutdown = self.task_client.fork("select");
 
-        // this should never get triggered
-        error!("cover traffic stream has been exhausted!")
+        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");
+        })
     }
 }

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

@@ -29,8 +29,6 @@ pub enum InputMessage {
         data: Vec<u8>,
         lane: TransmissionLane,
         max_retransmissions: Option<u32>,
-        // add trace_id for optional tracing of individual messages in debug mode
-        trace_id: Option<[u8; 12]>,
     },
 
     /// Creates a message used for a duplex anonymous communication where the recipient
@@ -47,7 +45,6 @@ pub enum InputMessage {
         reply_surbs: u32,
         lane: TransmissionLane,
         max_retransmissions: Option<u32>,
-        trace_id: Option<[u8; 12]>,
     },
 
     /// Attempt to use our internally received and stored `ReplySurb` to send the message back
@@ -93,14 +90,12 @@ impl InputMessage {
         data: Vec<u8>,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
-        trace_id: Option<[u8; 12]>,
     ) -> Self {
         let message = InputMessage::Regular {
             recipient,
             data,
             lane,
             max_retransmissions: None,
-            trace_id,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -115,7 +110,6 @@ impl InputMessage {
         reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
-        trace_id: Option<[u8; 12]>,
     ) -> Self {
         let message = InputMessage::Anonymous {
             recipient,
@@ -123,7 +117,6 @@ impl InputMessage {
             reply_surbs,
             lane,
             max_retransmissions: None,
-            trace_id,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -192,13 +185,4 @@ impl InputMessage {
         self.set_max_retransmissions(max_retransmissions);
         self
     }
-
-    pub fn trace_id(&self) -> Option<[u8; 12]> {
-        match self {
-            InputMessage::Regular { trace_id, .. } => *trace_id,
-            InputMessage::Anonymous { trace_id, .. } => *trace_id,
-            InputMessage::Premade { .. } | InputMessage::Reply { .. } => None,
-            InputMessage::MessageWrapper { message, .. } => message.trace_id(),
-        }
-    }
 }

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

@@ -2,9 +2,11 @@
 // 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::ShutdownToken;
+use nym_task::TaskClient;
 use tracing::*;
 use transceiver::ErasedGatewayError;
 
@@ -32,13 +34,13 @@ pub struct MixTrafficController {
     // in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
     consecutive_gateway_failure_count: usize,
 
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl MixTrafficController {
     pub fn new<T>(
         gateway_transceiver: T,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> (
         MixTrafficController,
         BatchMixMessageSender,
@@ -58,7 +60,7 @@ impl MixTrafficController {
                 mix_rx: message_receiver,
                 client_rx: client_receiver,
                 consecutive_gateway_failure_count: 0,
-                shutdown_token,
+                task_client,
             },
             message_sender,
             client_sender,
@@ -67,7 +69,7 @@ impl MixTrafficController {
 
     pub fn new_dynamic(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> (
         MixTrafficController,
         BatchMixMessageSender,
@@ -82,7 +84,7 @@ impl MixTrafficController {
                 mix_rx: message_receiver,
                 client_rx: client_receiver,
                 consecutive_gateway_failure_count: 0,
-                shutdown_token,
+                task_client,
             },
             message_sender,
             client_sender,
@@ -94,86 +96,72 @@ impl MixTrafficController {
         mut mix_packets: Vec<MixPacket>,
     ) -> Result<(), ErasedGatewayError> {
         debug_assert!(!mix_packets.is_empty());
-        let send_future = if mix_packets.len() == 1 {
-            // SAFETY: we just checked we have one packet
-            #[allow(clippy::unwrap_used)]
+
+        let result = if mix_packets.len() == 1 {
             let mix_packet = mix_packets.pop().unwrap();
-            self.gateway_transceiver.send_mix_packet(mix_packet)
+            self.gateway_transceiver.send_mix_packet(mix_packet).await
         } else {
-            self.gateway_transceiver.batch_send_mix_packets(mix_packets)
+            self.gateway_transceiver
+                .batch_send_mix_packets(mix_packets)
+                .await
         };
 
-        tokio::select! {
-            biased;
-            _ = self.shutdown_token.cancelled() => {
-                trace!("received shutdown while handling messages");
-                Ok(())
-            }
-            result = send_future => {
-                if result.is_err() {
-                    self.consecutive_gateway_failure_count += 1;
-                } else {
-                    trace!("We *might* have managed to forward sphinx packet(s) to the gateway!");
-                    self.consecutive_gateway_failure_count = 0;
-                }
-
-                result
-            }
+        if result.is_err() {
+            self.consecutive_gateway_failure_count += 1;
+        } else {
+            trace!("We *might* have managed to forward sphinx packet(s) to the gateway!");
+            self.consecutive_gateway_failure_count = 0;
         }
+
+        result
     }
 
-    async fn on_client_request(&mut self, client_request: ClientRequest) {
-        tokio::select! {
-            biased;
-             _ = self.shutdown_token.cancelled() => {
-                trace!("received shutdown while handling client request");
-            }
-            result = self.gateway_transceiver.send_client_request(client_request) => {
-                if let Err(err) = result {
-                    error!("Failed to send client request: {err}")
-                }
-            }
-        }
-    }
+    pub fn start(mut self) {
+        spawn_future(async move {
+            debug!("Started MixTrafficController with graceful shutdown support");
 
-    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;
+            while !self.task_client.is_shutdown() {
+                tokio::select! {
+                    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");
+                            break;
                         }
                     },
-                    None => {
-                        trace!("MixTrafficController: Stopping since channel closed");
+                    client_request = self.client_rx.recv() => match client_request {
+                        Some(client_request) => {
+                            match self.gateway_transceiver.send_client_request(client_request).await {
+                                Ok(_) => (),
+                                Err(e) => error!("Failed to send client request: {e}"),
+                            };
+                        },
+                        None => {
+                            tracing::trace!("MixTrafficController, client request channel closed");
+                        }
+                    },
+                    _ = self.task_client.recv() => {
+                        tracing::trace!("MixTrafficController: Received shutdown");
                         break;
                     }
-                },
-                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");
+            self.task_client.recv_timeout().await;
+
+            tracing::debug!("MixTrafficController: Exiting");
+        });
     }
 }

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

@@ -269,8 +269,6 @@ pub struct MockGateway {
 }
 
 impl Default for MockGateway {
-    // test code
-    #[allow(clippy::unwrap_used)]
     fn default() -> Self {
         MockGateway {
             dummy_identity: "3ebjp1Fb9hdcS1AR6AZihgeJiMHkB5jjJUsvqNnfQwU7"

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

@@ -10,17 +10,18 @@ use nym_sphinx::{
     acknowledgements::{identifier::recover_identifier, AckKey},
     chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID},
 };
-use nym_task::ShutdownToken;
+use nym_task::TaskClient;
 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(crate) struct AcknowledgementListener {
+pub(super) struct AcknowledgementListener {
     ack_key: Arc<AckKey>,
     ack_receiver: AcknowledgementReceiver,
     action_sender: AckActionSender,
     stats_tx: ClientStatsSender,
+    task_client: TaskClient,
 }
 
 impl AcknowledgementListener {
@@ -29,12 +30,14 @@ 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,
         }
     }
 
@@ -65,9 +68,14 @@ impl AcknowledgementListener {
         trace!("Received {frag_id} from the mix network");
         self.stats_tx
             .report(PacketStatisticsEvent::RealAckReceived(ack_content.len()).into());
-        let _ = self
+        if let Err(err) = self
             .action_sender
-            .unbounded_send(Action::new_remove(frag_id));
+            .unbounded_send(Action::new_remove(frag_id))
+        {
+            if !self.task_client.is_shutdown_poll() {
+                error!("Failed to send remove action to action controller: {err}");
+            }
+        }
     }
 
     async fn handle_ack_receiver_item(&mut self, item: Vec<Vec<u8>>) {
@@ -77,16 +85,11 @@ impl AcknowledgementListener {
         }
     }
 
-    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
+    pub(super) async fn run(&mut self) {
         debug!("Started AcknowledgementListener with graceful shutdown support");
 
-        loop {
+        while !self.task_client.is_shutdown() {
             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 => {
@@ -94,9 +97,12 @@ 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::ShutdownToken;
+use nym_task::TaskClient;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
@@ -82,7 +82,7 @@ impl Config {
     }
 }
 
-pub(crate) struct ActionController {
+pub(super) struct ActionController {
     /// Configurable parameters of the `ActionController`
     config: Config,
 
@@ -102,6 +102,8 @@ pub(crate) struct ActionController {
 
     /// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements.
     retransmission_sender: RetransmissionRequestSender,
+
+    task_client: TaskClient,
 }
 
 impl ActionController {
@@ -109,6 +111,7 @@ impl ActionController {
         config: Config,
         retransmission_sender: RetransmissionRequestSender,
         incoming_actions: AckActionReceiver,
+        task_client: TaskClient,
     ) -> Self {
         ActionController {
             config,
@@ -116,6 +119,7 @@ impl ActionController {
             pending_acks_timers: NonExhaustiveDelayQueue::new(),
             incoming_actions,
             retransmission_sender,
+            task_client,
         }
     }
 
@@ -190,11 +194,10 @@ impl ActionController {
         trace!("{frag_id} is updating its delay");
         // TODO: is it possible to solve this without either locking or temporarily removing the value?
         if let Some((pending_ack_data, queue_key)) = self.pending_acks_data.remove(&frag_id) {
-            // SAFETY: this Action is triggered by `RetransmissionRequestListener` (for 'normal' packets)
+            // this Action is triggered by `RetransmissionRequestListener` (for 'normal' packets)
             // or `ReplyController` (for 'reply' packets) which held the other potential
             // reference to this Arc. HOWEVER, before the Action was pushed onto the queue, the reference
             // was dropped hence this unwrap is safe.
-            #[allow(clippy::unwrap_used)]
             let mut inner_data = Arc::try_unwrap(pending_ack_data).unwrap();
             inner_data.update_retransmitted(delay);
 
@@ -206,7 +209,6 @@ impl ActionController {
     }
 
     // note: when the entry expires it's automatically removed from pending_acks_timers
-    #[allow(clippy::panic)]
     fn handle_expired_ack_timer(&mut self, expired_ack: Expired<FragmentIdentifier>) {
         let frag_id = expired_ack.into_inner();
 
@@ -222,9 +224,14 @@ 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)
-            let _ = self
+            if let Err(err) = self
                 .retransmission_sender
-                .unbounded_send(Arc::downgrade(pending_ack_data));
+                .unbounded_send(Arc::downgrade(pending_ack_data))
+            {
+                if !self.task_client.is_shutdown_poll() {
+                    tracing::error!("Failed to send pending ack for retransmission: {err}");
+                }
+            }
         } else {
             // this shouldn't cause any issues but shouldn't have happened to begin with!
             error!("An already removed pending ack has expired")
@@ -242,16 +249,11 @@ impl ActionController {
         }
     }
 
-    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
+    pub(super) async fn run(&mut self) {
         debug!("Started ActionController with graceful shutdown support");
 
-        loop {
+        while !self.task_client.is_shutdown() {
             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 => {
@@ -268,8 +270,13 @@ 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,20 +10,21 @@ 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::ShutdownToken;
+use nym_task::TaskClient;
 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(crate) struct InputMessageListener<R>
+pub(super) 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>
@@ -37,11 +38,13 @@ 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,
         }
     }
 
@@ -65,12 +68,16 @@ where
         max_retransmissions: Option<u32>,
     ) {
         // offload reply handling to the dedicated task
-        let _ =
+        if let Err(err) =
             self.reply_controller_sender
-                .send_reply(recipient_tag, data, lane, max_retransmissions);
+                .send_reply(recipient_tag, data, lane, max_retransmissions)
+        {
+            if !self.task_client.is_shutdown_poll() {
+                error!("failed to send a reply - {err}");
+            }
+        }
     }
 
-    #[instrument(skip_all)]
     async fn handle_plain_message(
         &mut self,
         recipient: Recipient,
@@ -78,18 +85,16 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        trace_id: Option<[u8; 12]>,
     ) {
         if let Err(err) = self
             .message_handler
-            .try_send_plain_message(recipient, content, lane, packet_type, max_retransmissions, trace_id)
+            .try_send_plain_message(recipient, content, lane, packet_type, max_retransmissions)
             .await
         {
             warn!("failed to send a plain message - {err}")
         }
     }
 
-    #[instrument(skip_all)]
     async fn handle_repliable_message(
         &mut self,
         recipient: Recipient,
@@ -98,7 +103,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        trace_id: Option<[u8; 12]>,
     ) {
         if let Err(err) = self
             .message_handler
@@ -109,7 +113,6 @@ where
                 lane,
                 packet_type,
                 max_retransmissions,
-                trace_id,
             )
             .await
         {
@@ -117,30 +120,20 @@ where
         }
     }
 
-    #[allow(clippy::panic)]
-    #[instrument(skip_all)]
     async fn on_input_message(&mut self, msg: InputMessage) {
-        let trace_id = msg.trace_id();
-        if let Some(tid) = trace_id {
-            tracing::warn!("Processing input message with trace_id: {:?}", tid);
-        }
-
         match msg {
             InputMessage::Regular {
                 recipient,
                 data,
                 lane,
                 max_retransmissions,
-                ..
             } => {
-                warn!("Handling regular input message with trace_id: {:?}", trace_id);
                 self.handle_plain_message(
                     recipient,
                     data,
                     lane,
                     PacketType::Mix,
                     max_retransmissions,
-                    trace_id
                 )
                 .await
             }
@@ -150,9 +143,7 @@ where
                 reply_surbs,
                 lane,
                 max_retransmissions,
-                ..
             } => {
-                warn!("Handling anonymous input message with trace_id: {:?}", trace_id);
                 self.handle_repliable_message(
                     recipient,
                     data,
@@ -160,7 +151,6 @@ where
                     lane,
                     PacketType::Mix,
                     max_retransmissions,
-                    trace_id
                 )
                 .await
             }
@@ -170,7 +160,6 @@ where
                 lane,
                 max_retransmissions,
             } => {
-                warn!("Handling reply input message with trace_id: {:?}", trace_id);
                 self.handle_reply(recipient_tag, data, lane, max_retransmissions)
                     .await;
             }
@@ -184,16 +173,13 @@ where
                     data,
                     lane,
                     max_retransmissions,
-                    ..
                 } => {
-                    tracing::warn!("Handling regular input message with trace_id: {:?}", trace_id);
                     self.handle_plain_message(
                         recipient,
                         data,
                         lane,
                         packet_type,
                         max_retransmissions,
-                        trace_id
                     )
                     .await
                 }
@@ -203,7 +189,6 @@ where
                     reply_surbs,
                     lane,
                     max_retransmissions,
-                    ..
                 } => {
                     self.handle_repliable_message(
                         recipient,
@@ -212,7 +197,6 @@ where
                         lane,
                         packet_type,
                         max_retransmissions,
-                        trace_id
                     )
                     .await
                 }
@@ -229,24 +213,16 @@ where
                     self.handle_premade_packets(msgs, lane).await
                 }
                 // MessageWrappers can't be nested
-                InputMessage::MessageWrapper { .. } => {
-                    panic!("attempted to use nested MessageWrapper")
-                }
+                InputMessage::MessageWrapper { .. } => unimplemented!(),
             },
         };
     }
 
-    #[instrument(skip_all)]
-    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
+    pub(super) async fn run(&mut self) {
         debug!("Started InputMessageListener with graceful shutdown support");
 
-        loop {
+        while !self.task_client.is_shutdown() {
             tokio::select! {
-                biased;
-                _ = shutdown_token.cancelled() => {
-                    tracing::trace!("InputMessageListener: Received shutdown");
-                    break;
-                }
                 input_msg = self.input_receiver.recv() => match input_msg {
                     Some(input_msg) => {
                         self.on_input_message(input_msg).await;
@@ -256,9 +232,12 @@ where
                         break;
                     }
                 },
-
+                _ = self.task_client.recv() => {
+                    tracing::trace!("InputMessageListener: Received shutdown");
+                }
             }
         }
+        self.task_client.recv_timeout().await;
         tracing::debug!("InputMessageListener: Exiting");
     }
 }

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

@@ -10,6 +10,7 @@ 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;
@@ -22,11 +23,13 @@ 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};
 
@@ -187,9 +190,6 @@ 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,14 +197,12 @@ 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,
         }
     }
 
@@ -214,7 +212,7 @@ impl Config {
     }
 }
 
-pub(crate) struct AcknowledgementController<R>
+pub(super) struct AcknowledgementController<R>
 where
     R: CryptoRng + Rng,
 {
@@ -236,6 +234,7 @@ where
         message_handler: MessageHandler<R>,
         reply_controller_sender: ReplyControllerSender,
         stats_tx: ClientStatsSender,
+        task_client: TaskClient,
     ) -> Self {
         let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
 
@@ -245,6 +244,7 @@ where
             action_config,
             retransmission_tx,
             connectors.ack_action_receiver,
+            task_client.fork("action_controller"),
         );
 
         // will listen for any acks coming from the network
@@ -253,6 +253,7 @@ 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
@@ -260,6 +261,7 @@ 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
@@ -269,13 +271,16 @@ where
             message_handler,
             retransmission_rx,
             reply_controller_sender,
-            config.packet_type,
+            task_client.fork("retransmission_request_listener"),
         );
 
         // 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);
+        let sent_notification_listener = SentNotificationListener::new(
+            connectors.sent_notifier,
+            connectors.ack_action_sender,
+            task_client.with_suffix("sent_notification_listener"),
+        );
 
         AcknowledgementController {
             acknowledgement_listener,
@@ -286,21 +291,36 @@ where
         }
     }
 
-    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,
-        )
+    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!");
+        });
+
+        spawn_future(async move {
+            input_message_listener.run().await;
+            debug!("The input listener has finished execution!");
+        });
+
+        spawn_future(async move {
+            retransmission_request_listener.run(packet_type).await;
+            debug!("The retransmission request listener has finished execution!");
+        });
+
+        spawn_future(async move {
+            sent_notification_listener.run().await;
+            debug!("The sent notification listener has finished execution!");
+        });
+
+        spawn_future(async move {
+            action_controller.run().await;
+            debug!("The controller has finished execution!");
+        });
     }
 }

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, ShutdownToken};
+use nym_task::{connections::TransmissionLane, TaskClient};
 use rand::{CryptoRng, Rng};
 use std::sync::{Arc, Weak};
 use tracing::*;
 
 // responsible for packet retransmission upon fired timer
-pub(crate) struct RetransmissionRequestListener<R> {
+pub(super) struct RetransmissionRequestListener<R> {
     maximum_retransmissions: Option<u32>,
     action_sender: AckActionSender,
     message_handler: MessageHandler<R>,
     request_receiver: RetransmissionRequestReceiver,
     reply_controller_sender: ReplyControllerSender,
-    packet_type: PacketType,
+    task_client: TaskClient,
 }
 
 impl<R> RetransmissionRequestListener<R>
@@ -38,7 +38,7 @@ where
         message_handler: MessageHandler<R>,
         request_receiver: RetransmissionRequestReceiver,
         reply_controller_sender: ReplyControllerSender,
-        packet_type: PacketType,
+        task_client: TaskClient,
     ) -> Self {
         RetransmissionRequestListener {
             maximum_retransmissions,
@@ -46,7 +46,7 @@ where
             message_handler,
             request_receiver,
             reply_controller_sender,
-            packet_type,
+            task_client,
         }
     }
 
@@ -60,13 +60,14 @@ where
 
         // TODO: Figure out retransmission packet type signaling
         self.message_handler
-            .try_prepare_single_chunk_for_sending(packet_recipient, chunk_data, packet_type, None)
+            .try_prepare_single_chunk_for_sending(packet_recipient, chunk_data, packet_type)
             .await
     }
 
     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,
@@ -96,18 +97,22 @@ where
             } => {
                 // if this is retransmission for reply, offload it to the dedicated task
                 // that deals with all the surbs
-                let _ = self.reply_controller_sender.send_retransmission_data(
+                if let Err(err) = 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(),
-                    self.packet_type,
+                    packet_type,
                 )
                 .await
             }
@@ -148,9 +153,14 @@ 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`
-        let _ = self
+        if let Err(err) = self
             .action_sender
-            .unbounded_send(Action::new_update_pending_ack(frag_id, new_delay));
+            .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}");
+            }
+        }
 
         // send to `OutQueueControl` to eventually send to the mix network
         self.message_handler
@@ -164,26 +174,24 @@ where
             .await
     }
 
-    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
+    pub(super) async fn run(&mut self, packet_type: PacketType) {
         debug!("Started RetransmissionRequestListener with graceful shutdown support");
 
-        loop {
+        while !self.task_client.is_shutdown() {
             tokio::select! {
-                biased;
-                _ = 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).await,
+                    Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack, packet_type).await,
                     None => {
                         tracing::trace!("RetransmissionRequestListener: Stopping since channel closed");
                         break;
                     }
                 },
-
+                _ = self.task_client.recv() => {
+                    tracing::trace!("RetransmissionRequestListener: Received shutdown");
+                }
             }
         }
+        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,25 +5,29 @@ 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(crate) struct SentNotificationListener {
+pub(super) 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,
         }
     }
 
@@ -32,18 +36,37 @@ impl SentNotificationListener {
             trace!("sent off a cover message - no need to start retransmission timer!");
             return;
         }
-        let _ = self
+        if let Err(err) = self
             .action_sender
-            .unbounded_send(Action::new_start_timer(frag_id));
+            .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}");
+            }
+        }
     }
 
-    pub(crate) async fn run(&mut self) {
+    pub(super) async fn run(&mut self) {
         debug!("Started SentNotificationListener with graceful shutdown support");
 
-        while let Some(frag_id) = self.sent_notifier.next().await {
-            self.on_sent_message(frag_id).await;
+        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;
+                }
+            }
         }
-
+        assert!(self.task_client.is_shutdown_poll());
         tracing::debug!("SentNotificationListener: Exiting");
     }
 }

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

@@ -20,14 +20,14 @@ use nym_sphinx::params::{PacketSize, PacketType};
 use nym_sphinx::preparer::{MessagePreparer, PreparedFragment};
 use nym_sphinx::Delay;
 use nym_task::connections::TransmissionLane;
-use nym_task::ShutdownToken;
+use nym_task::TaskClient;
 use nym_topology::{NymRouteProvider, NymTopologyError};
 use rand::{CryptoRng, Rng};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 use thiserror::Error;
-use tracing::{debug, error, info, instrument, trace, warn};
+use tracing::{debug, error, info, trace, warn};
 
 // TODO: move that error elsewhere since it seems to be contaminating different files
 #[derive(Debug, Error)]
@@ -35,9 +35,6 @@ pub enum PreparationError {
     #[error(transparent)]
     NymTopologyError(#[from] NymTopologyError),
 
-    #[error("message wasn't split into any fragments!")]
-    EmptyFragments,
-
     #[error("message too long for a single SURB, splitting into {fragments} fragments.")]
     MessageTooLongForSingleSurb { fragments: usize },
 
@@ -189,7 +186,7 @@ pub(crate) struct MessageHandler<R> {
     topology_access: TopologyAccessor,
     reply_key_storage: SentReplyKeys,
     tag_storage: UsedSenderTags,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl<R> MessageHandler<R>
@@ -205,7 +202,7 @@ where
         topology_access: TopologyAccessor,
         reply_key_storage: SentReplyKeys,
         tag_storage: UsedSenderTags,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self
     where
         R: Copy,
@@ -228,7 +225,7 @@ where
             topology_access,
             reply_key_storage,
             tag_storage,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -323,16 +320,6 @@ where
             });
         }
 
-        if fragment.is_empty() {
-            error!("CRITICAL FAILURE: our split message didn't result in any sendable fragments");
-            return Err(SurbWrappedPreparationError {
-                source: PreparationError::EmptyFragments,
-                returned_surbs: Some(vec![reply_surb]),
-            });
-        }
-
-        // SAFETY: we just checked we have one fragment
-        #[allow(clippy::unwrap_used)]
         let chunk = fragment.pop().unwrap();
         let chunk_clone = chunk.clone();
         let prepared_fragment = self
@@ -476,7 +463,6 @@ where
         self.forward_messages(msgs, lane).await;
     }
 
-    #[instrument(skip_all)]
     pub(crate) async fn try_send_plain_message(
         &mut self,
         recipient: Recipient,
@@ -484,7 +470,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        trace_id: Option<[u8; 12]>,
     ) -> Result<(), PreparationError> {
         let message = NymMessage::new_plain(message);
         self.try_split_and_send_non_reply_message(
@@ -493,12 +478,10 @@ where
             lane,
             packet_type,
             max_retransmissions,
-            trace_id,
         )
         .await
     }
 
-    #[instrument(skip_all)]
     pub(crate) async fn try_split_and_send_non_reply_message(
         &mut self,
         message: NymMessage,
@@ -506,7 +489,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        trace_id: Option<[u8; 12]>,
     ) -> Result<(), PreparationError> {
         debug!("Sending non-reply message with packet type {packet_type}");
         // TODO: I really dislike existence of this assertion, it implies code has to be re-organised
@@ -539,7 +521,6 @@ where
                 &self.config.ack_key,
                 &recipient,
                 packet_type,
-                trace_id
             )?;
 
             let real_message = RealMessage::new(
@@ -554,7 +535,6 @@ where
             pending_acks.push(pending_ack);
         }
 
-        drop(topology_permit);
         self.insert_pending_acks(pending_acks);
         self.forward_messages(real_messages, lane).await;
 
@@ -591,7 +571,6 @@ where
             TransmissionLane::AdditionalReplySurbs,
             packet_type,
             max_retransmissions,
-            None,
         )
         .await?;
 
@@ -609,7 +588,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        trace_id: Option<[u8; 12]>,
     ) -> Result<(), SurbWrappedPreparationError> {
         debug!("Sending message with reply SURBs with packet type {packet_type}");
         let sender_tag = self.get_or_create_sender_tag(&recipient);
@@ -633,7 +611,6 @@ where
             lane,
             packet_type,
             max_retransmissions,
-            trace_id,
         )
         .await?;
 
@@ -648,7 +625,6 @@ where
         recipient: Recipient,
         chunk: Fragment,
         packet_type: PacketType,
-        trace_id: Option<[u8; 12]>,
     ) -> Result<PreparedFragment, PreparationError> {
         debug!("Sending single chunk with packet type {packet_type}");
         let topology_permit = self.topology_access.get_read_permit().await;
@@ -660,7 +636,6 @@ where
             &self.config.ack_key,
             &recipient,
             packet_type,
-            trace_id,
         )?;
 
         Ok(prepared_fragment)
@@ -682,7 +657,6 @@ where
             .zip(reply_surbs.into_iter())
             .map(|(fragment, reply_surb)| {
                 // unwrap here is fine as we know we have a valid topology
-                #[allow(clippy::unwrap_used)]
                 self.message_preparer
                     .prepare_reply_chunk_for_sending(
                         fragment,
@@ -723,7 +697,7 @@ where
             .action_sender
             .unbounded_send(Action::UpdatePendingAck(id, new_delay))
         {
-            if !self.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send update action to the controller: {err}");
             }
         }
@@ -734,7 +708,7 @@ where
             .action_sender
             .unbounded_send(Action::new_insert(pending_acks))
         {
-            if !self.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send insert action to the controller: {err}");
             }
         }
@@ -742,21 +716,17 @@ where
 
     // tells real message sender (with the poisson timer) to send this to the mix network
     pub(crate) async fn forward_messages(
-        &mut self,
+        &self,
         messages: Vec<RealMessage>,
         transmission_lane: TransmissionLane,
     ) {
-        tokio::select! {
-            biased;
-            _ = self.shutdown_token.cancelled() => {
-                trace!("received shutdown while attempting to forward mixnet messages");
-            }
-            sending_res = self.real_message_sender.send((messages, transmission_lane)) => {
-                if sending_res.is_err() {
-                    error!(
-                        "failed to forward mixnet messages due to closed channel (outside of shutdown!)"
-                    );
-                }
+        if let Err(err) = self
+            .real_message_sender
+            .send((messages, transmission_lane))
+            .await
+        {
+            if !self.task_client.is_shutdown_poll() {
+                error!("Failed to forward messages to the real message sender: {err}");
             }
         }
     }

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

@@ -14,21 +14,26 @@ use crate::client::replies::reply_controller::{
     ReplyController, ReplyControllerReceiver, ReplyControllerSender,
 };
 use crate::client::replies::reply_storage::CombinedReplyStorage;
-use crate::client::{
-    inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
-    real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
-    topology_control::TopologyAccessor,
-};
 use crate::config;
+use crate::{
+    client::{
+        inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
+        real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
+        topology_control::TopologyAccessor,
+    },
+    spawn_future,
+};
 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::ShutdownToken;
+use nym_task::TaskClient;
 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};
@@ -64,7 +69,6 @@ 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)
     }
@@ -142,7 +146,7 @@ impl RealMessagesController<OsRng> {
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
         stats_tx: ClientStatsSender,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         let rng = OsRng;
 
@@ -174,7 +178,7 @@ impl RealMessagesController<OsRng> {
             topology_access.clone(),
             reply_storage.key_storage(),
             reply_storage.tags_storage(),
-            shutdown_token.clone(),
+            task_client.fork("message_handler"),
         );
 
         let ack_control = AcknowledgementController::new(
@@ -184,6 +188,7 @@ impl RealMessagesController<OsRng> {
             message_handler.clone(),
             reply_controller_sender,
             stats_tx.clone(),
+            task_client.fork("ack_control"),
         );
 
         let reply_control = ReplyController::new(
@@ -191,6 +196,7 @@ impl RealMessagesController<OsRng> {
             message_handler,
             reply_storage,
             reply_controller_receiver,
+            task_client.fork("reply_controller"),
         );
 
         let out_queue_control = OutQueueControl::new(
@@ -203,7 +209,7 @@ impl RealMessagesController<OsRng> {
             lane_queue_lengths,
             client_connection_rx,
             stats_tx,
-            shutdown_token.clone(),
+            task_client.with_suffix("out_queue_control"),
         );
 
         RealMessagesController {
@@ -213,13 +219,20 @@ impl RealMessagesController<OsRng> {
         }
     }
 
-    pub fn into_tasks(
-        self,
-    ) -> (
-        OutQueueControl<OsRng>,
-        ReplyController<OsRng>,
-        AcknowledgementController<OsRng>,
-    ) {
-        (self.out_queue_control, self.reply_control, self.ack_control)
+    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!");
+        });
+        spawn_future(async move {
+            reply_control.run().await;
+            debug!("The reply controller has finished execution!");
+        });
+
+        ack_control.start(packet_type);
     }
 }

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::ShutdownToken;
+use nym_task::TaskClient;
 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,
 
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 #[derive(Debug)]
@@ -179,7 +179,7 @@ where
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
         stats_tx: ClientStatsSender,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         OutQueueControl {
             config,
@@ -194,7 +194,7 @@ where
             client_connection_rx,
             lane_queue_lengths,
             stats_tx,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -249,8 +249,6 @@ where
                     }
                 };
 
-                // SAFETY: our topology must be valid at this point
-                #[allow(clippy::expect_used)]
                 (
                     generate_loop_cover_packet(
                         &mut self.rng,
@@ -280,33 +278,17 @@ where
             }
         };
 
-        let sending_res = tokio::select! {
-            biased;
-            _ = self.shutdown_token.cancelled() => {
-                trace!("received shutdown signal while attempting to send mix message");
-                return
-            }
-            sending_res = self.mix_tx.send(vec![next_message]) => {
-                sending_res
-            }
-        };
-
-        match sending_res {
-            Err(_) => {
-                if !self.shutdown_token.is_cancelled() {
-                    tracing::error!(
-                        "failed to send mixnet packet due to closed channel (outside of shutdown!)"
-                    );
-                }
-            }
-            Ok(_) => {
-                let event = if fragment_id.is_some() {
-                    PacketStatisticsEvent::RealPacketSent(packet_size)
-                } else {
-                    PacketStatisticsEvent::CoverPacketSent(packet_size)
-                };
-                self.stats_tx.report(event.into());
+        if let Err(err) = self.mix_tx.send(vec![next_message]).await {
+            if !self.task_client.is_shutdown_poll() {
+                tracing::error!("Failed to send: {err}");
             }
+        } else {
+            let event = if fragment_id.is_some() {
+                PacketStatisticsEvent::RealPacketSent(packet_size)
+            } else {
+                PacketStatisticsEvent::CoverPacketSent(packet_size)
+            };
+            self.stats_tx.report(event.into());
         }
 
         // notify ack controller about sending our message only after we actually managed to push it
@@ -457,8 +439,6 @@ where
                     tracing::trace!("handling real_messages: size: {}", real_messages.len());
 
                     self.transmission_buffer.store(&conn_id, real_messages);
-                    // SAFETY: we just stored the message
-                    #[allow(clippy::expect_used)]
                     let real_next = self.pop_next_message().expect("Just stored one");
 
                     Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
@@ -507,8 +487,6 @@ where
 
                 // First store what we got for the given connection id
                 self.transmission_buffer.store(&conn_id, real_messages);
-                // SAFETY: we just stored the message
-                #[allow(clippy::expect_used)]
                 let real_next = self.pop_next_message().expect("we just added one");
 
                 Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
@@ -536,7 +514,9 @@ where
     }
 
     #[cfg(not(target_arch = "wasm32"))]
-    fn log_status(&self) {
+    fn log_status(&self, shutdown: &mut TaskClient) {
+        use crate::error::ClientCoreStatusMessage;
+
         let packets = self.transmission_buffer.total_size();
         let lanes = self.transmission_buffer.lanes();
         let mult = self.sending_delay_controller.current_multiplier();
@@ -565,33 +545,32 @@ where
             tracing::debug!("{status_str}");
         }
 
-        // 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));
-        // }
+        // 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(crate) async fn run(&mut self) {
+    pub(super) async fn run(&mut self) {
         debug!("Started OutQueueControl with graceful shutdown support");
 
-        // avoid borrow on self
-        let shutdown_token = self.shutdown_token.clone();
+        let mut shutdown = self.task_client.fork("select");
+
         #[cfg(not(target_arch = "wasm32"))]
         {
             let mut status_timer = tokio::time::interval(Duration::from_secs(5));
 
-            loop {
+            while !shutdown.is_shutdown() {
                 tokio::select! {
                     biased;
-                    _ = shutdown_token.cancelled() => {
+                    _ = shutdown.recv() => {
                         tracing::trace!("OutQueueControl: Received shutdown");
                         break;
                     }
                     _ = status_timer.tick() => {
-                        self.log_status();
+                        self.log_status(&mut shutdown);
                     }
                     next_message = self.next() => if let Some(next_message) = next_message {
                         self.on_message(next_message).await;
@@ -601,16 +580,16 @@ where
                     }
                 }
             }
+            shutdown.recv_timeout().await;
         }
 
         #[cfg(target_arch = "wasm32")]
         {
-            loop {
+            while !shutdown.is_shutdown() {
                 tokio::select! {
                     biased;
-                    _ = shutdown_token.cancelled() => {
+                    _ = shutdown.recv() => {
                         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,13 +83,11 @@ 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::ShutdownToken;
+use nym_task::TaskClient;
 use std::collections::HashSet;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 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: crate::client::helpers::Instant,
+    last_stale_check: Instant,
 }
 
 impl<R: MessageReceiver> ReceivedMessagesBufferInner<R> {
@@ -154,7 +154,7 @@ impl<R: MessageReceiver> ReceivedMessagesBufferInner<R> {
     }
 
     fn cleanup_stale_buffers(&mut self) {
-        let now = get_time_now();
+        let now = Instant::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,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 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,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         ReceivedMessagesBuffer {
             inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
@@ -190,15 +190,14 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                 message_sender: None,
                 recently_reconstructed: HashSet::new(),
                 stats_tx,
-                last_stale_check: get_time_now(),
+                last_stale_check: Instant::now(),
             })),
             reply_key_storage,
             reply_controller_sender,
-            shutdown_token,
+            task_client,
         }
     }
 
-    #[allow(clippy::panic)]
     async fn disconnect_sender(&mut self) {
         let mut guard = self.inner.lock().await;
         if guard.message_sender.is_none() {
@@ -209,7 +208,6 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
         guard.message_sender = None;
     }
 
-    #[allow(clippy::panic)]
     async fn connect_sender(&mut self, sender: ReconstructedMessagesSender) {
         let mut guard = self.inner.lock().await;
         if guard.message_sender.is_some() {
@@ -315,7 +313,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                     reply_surbs,
                     from_surb_request,
                 ) {
-                    if !self.shutdown_token.is_cancelled() {
+                    if !self.task_client.is_shutdown_poll() {
                         error!("{err}");
                     }
                 }
@@ -338,7 +336,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                         .reply_controller_sender
                         .send_additional_surbs_request(*recipient, amount)
                     {
-                        if !self.shutdown_token.is_cancelled() {
+                        if !self.task_client.is_shutdown_poll() {
                             error!("{err}");
                         }
                     }
@@ -465,22 +463,22 @@ pub enum ReceivedBufferMessage {
     ReceiverDisconnect,
 }
 
-pub(crate) struct RequestReceiver<R: MessageReceiver> {
+struct RequestReceiver<R: MessageReceiver> {
     received_buffer: ReceivedMessagesBuffer<R>,
     query_receiver: ReceivedBufferRequestReceiver,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl<R: MessageReceiver> RequestReceiver<R> {
     fn new(
         received_buffer: ReceivedMessagesBuffer<R>,
         query_receiver: ReceivedBufferRequestReceiver,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         RequestReceiver {
             received_buffer,
             query_receiver,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -495,70 +493,66 @@ impl<R: MessageReceiver> RequestReceiver<R> {
         }
     }
 
-    pub(crate) async fn run(&mut self) {
+    async fn run(&mut self) {
         debug!("Started RequestReceiver with graceful shutdown support");
-        loop {
+        while !self.task_client.is_shutdown() {
             tokio::select! {
                 biased;
-                _ = self.shutdown_token.cancelled() => {
+                _ = self.task_client.recv() => {
                     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");
     }
 }
 
-pub(crate) struct FragmentedMessageReceiver<R: MessageReceiver> {
+struct FragmentedMessageReceiver<R: MessageReceiver> {
     received_buffer: ReceivedMessagesBuffer<R>,
     mixnet_packet_receiver: MixnetMessageReceiver,
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 impl<R: MessageReceiver> FragmentedMessageReceiver<R> {
     fn new(
         received_buffer: ReceivedMessagesBuffer<R>,
         mixnet_packet_receiver: MixnetMessageReceiver,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         FragmentedMessageReceiver {
             received_buffer,
             mixnet_packet_receiver,
-            shutdown_token,
+            task_client,
         }
     }
 
-    pub(crate) async fn run(&mut self) -> Result<(), MessageRecoveryError> {
+    async fn run(&mut self) -> Result<(), MessageRecoveryError> {
         debug!("Started FragmentedMessageReceiver with graceful shutdown support");
-        loop {
+        while !self.task_client.is_shutdown() {
             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(())
     }
@@ -577,31 +571,42 @@ impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferControll
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
         metrics_reporter: ClientStatsSender,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         let received_buffer = ReceivedMessagesBuffer::new(
             local_encryption_keypair,
             reply_key_storage,
             reply_controller_sender,
             metrics_reporter,
-            shutdown_token.clone(),
+            task_client.fork("received_messages_buffer"),
         );
 
         ReceivedMessagesBufferController {
             fragmented_message_receiver: FragmentedMessageReceiver::new(
                 received_buffer.clone(),
                 mixnet_packet_receiver,
-                shutdown_token.clone(),
+                task_client.fork("fragmented_message_receiver"),
             ),
             request_receiver: RequestReceiver::new(
                 received_buffer,
                 query_receiver,
-                shutdown_token.clone(),
+                task_client.with_suffix("request_receiver"),
             ),
         }
     }
 
-    pub(crate) fn into_tasks(self) -> (FragmentedMessageReceiver<R>, RequestReceiver<R>) {
-        (self.fragmented_message_receiver, self.request_receiver)
+    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}"),
+            }
+        });
+        spawn_future(async move {
+            request_receiver.run().await;
+        });
     }
 }

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::ShutdownToken;
+use nym_task::TaskClient;
 use rand::rngs::OsRng;
 use rand::{CryptoRng, Rng};
 use std::time::Duration;
@@ -60,6 +60,9 @@ pub struct ReplyController<R> {
     receiver_controller: ReceiverReplyController<R>,
 
     request_receiver: ReplyControllerReceiver,
+
+    // Listen for shutdown signals
+    task_client: TaskClient,
 }
 
 impl ReplyController<OsRng> {
@@ -68,6 +71,7 @@ impl ReplyController<OsRng> {
         message_handler: MessageHandler<OsRng>,
         full_reply_storage: CombinedReplyStorage,
         request_receiver: ReplyControllerReceiver,
+        task_client: TaskClient,
     ) -> Self {
         ReplyController {
             config,
@@ -82,6 +86,7 @@ impl ReplyController<OsRng> {
                 message_handler,
             ),
             request_receiver,
+            task_client,
         }
     }
 }
@@ -143,21 +148,22 @@ where
         self.sender_controller.inspect_and_clear_stale_data(now)
     }
 
-    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
+    pub(crate) async fn run(&mut self) {
         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);
 
-        loop {
+        while !shutdown.is_shutdown() {
             tokio::select! {
                 biased;
-                _ = shutdown_token.cancelled() => {
+                _ = shutdown.recv() => {
                     tracing::trace!("ReplyController: Received shutdown");
-                    break;
                 },
                 req = self.request_receiver.next() => match req {
                     Some(req) => self.handle_request(req).await,
@@ -175,6 +181,7 @@ where
                 }
             }
         }
+        assert!(shutdown.is_shutdown_poll());
         tracing::debug!("ReplyController: Exiting");
     }
 }

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

@@ -155,9 +155,8 @@ where
         data: Vec<Arc<PendingAcknowledgement>>,
     ) {
         trace!("re-inserting pending retransmissions for {recipient}");
-        // SAFETY: the underlying entry MUST exist as we've just got data from there
+        // the underlying entry MUST exist as we've just got data from there
         // and we hold a mut reference
-        #[allow(clippy::expect_used)]
         let map_entry = &mut self
             .surb_senders
             .get_mut(recipient)
@@ -430,7 +429,6 @@ where
             .pop_at_most_n_next_messages_at_random(amount)
     }
 
-    #[allow(clippy::panic)]
     async fn try_clear_pending_queue(&mut self, target: AnonymousSenderTag) {
         trace!("trying to clear pending queue");
         let available_surbs = self.surbs_storage.available_surbs(&target);

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

@@ -16,17 +16,21 @@
 #![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, ShutdownToken, ShutdownTracker};
+use nym_task::{connections::TransmissionLane, TaskClient};
 use std::time::Duration;
 
-/// Time interval between reporting statistics locally (logging/shutdown_token)
+use crate::{
+    client::inbound_messages::{InputMessage, InputMessageSender},
+    spawn_future,
+};
+
+/// Time interval between reporting statistics locally (logging/task_client)
 const LOCAL_REPORT_INTERVAL: Duration = Duration::from_secs(2);
 /// Interval for taking snapshots of the statistics
 const SNAPSHOT_INTERVAL: Duration = Duration::from_millis(500);
@@ -47,6 +51,9 @@ 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 {
@@ -55,20 +62,24 @@ impl StatisticsControl {
         client_type: String,
         client_stats_id: String,
         report_tx: InputMessageSender,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> (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), shutdown_token),
+            ClientStatsSender::new(Some(stats_tx), task_client_stats_sender),
         )
     }
 
@@ -80,7 +91,6 @@ impl StatisticsControl {
             stats_report.into(),
             TransmissionLane::General,
             None,
-            None,
         );
         if let Err(err) = self.report_tx.send(report_message).await {
             tracing::error!("Failed to report client stats: {err:?}");
@@ -89,8 +99,7 @@ impl StatisticsControl {
         }
     }
 
-    // manually control the shutdown mechanism as we don't want to get interrupted mid-snapshot
-    pub async fn run(&mut self, shutdown_token: ShutdownToken) {
+    async fn run(&mut self) {
         tracing::debug!("Started StatisticsControl with graceful shutdown support");
 
         #[cfg(not(target_arch = "wasm32"))]
@@ -120,10 +129,10 @@ impl StatisticsControl {
         let mut snapshot_interval =
             gloo_timers::future::IntervalStream::new(SNAPSHOT_INTERVAL.as_millis() as u32);
 
-        loop {
+        while !self.task_client.is_shutdown() {
             tokio::select! {
                 biased;
-                _ = shutdown_token.cancelled() => {
+                _ = self.task_client.recv() => {
                     tracing::trace!("StatisticsControl: Received shutdown");
                     break;
                 },
@@ -148,34 +157,34 @@ impl StatisticsControl {
                 }
 
                 _ = local_report_interval.next() => {
-                    self.stats.local_report();
+                    self.stats.local_report(&mut self.task_client);
                 }
             }
         }
         tracing::debug!("StatisticsControl: Exiting");
     }
 
+    pub(crate) fn start(mut self) {
+        spawn_future(async move {
+            self.run().await;
+        })
+    }
+
     pub(crate) fn create_and_start(
         reporting_config: StatsReporting,
         client_type: String,
         client_stats_id: String,
         report_tx: InputMessageSender,
-        shutdown_tracker: &ShutdownTracker,
+        task_client: TaskClient,
     ) -> ClientStatsSender {
-        let (mut controller, sender) = Self::create(
+        let (controller, sender) = Self::create(
             reporting_config,
             client_type,
             client_stats_id,
             report_tx,
-            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",
+            task_client,
         );
+        controller.start();
         sender
     }
 }

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

@@ -126,7 +126,7 @@ impl TopologyAccessor {
             .map(|p| p.topology.clone())
     }
 
-    pub async fn current_route_provider(&self) -> Option<RwLockReadGuard<'_, NymRouteProvider>> {
+    pub async fn current_route_provider(&self) -> Option<RwLockReadGuard<NymRouteProvider>> {
         let provider = self.inner.topology.read().await;
         if provider.topology.is_empty() {
             None

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

@@ -1,9 +1,11 @@
 // 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::*;
@@ -39,6 +41,8 @@ pub struct TopologyRefresher {
 
     refresh_rate: Duration,
     consecutive_failure_count: usize,
+
+    task_client: TaskClient,
 }
 
 impl TopologyRefresher {
@@ -46,12 +50,14 @@ 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,
         }
     }
 
@@ -138,30 +144,37 @@ impl TopologyRefresher {
         }
     }
 
-    // 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");
+    pub fn start(mut self) {
+        spawn_future(async move {
+            debug!("Started TopologyRefresher with graceful shutdown support");
 
-        #[cfg(not(target_arch = "wasm32"))]
-        let mut interval =
-            tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(self.refresh_rate));
+            #[cfg(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 interval.next().await.is_some() {
-            self.try_refresh().await;
-        }
-
-        // this should never get triggered
-        error!("topology refresher interval has been exhausted!")
+            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");
+        })
     }
 }

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

@@ -2,10 +2,8 @@
 // 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;
@@ -41,43 +39,30 @@ impl Config {
 pub struct NymApiTopologyProvider {
     config: Config,
 
-    validator_client: nym_http_api_client::Client,
+    validator_client: nym_validator_client::client::NymApiClient,
     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>,
-        validator_client: nym_http_api_client::Client,
+        mut validator_client: nym_validator_client::client::NymApiClient,
     ) -> Self {
         nym_api_urls.shuffle(&mut thread_rng());
-        let mut provider = NymApiTopologyProvider {
+        validator_client.change_nym_api(nym_api_urls[0].clone());
+
+        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.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");
+        self.validator_client.use_bincode = false;
     }
 
     fn use_next_nym_api(&mut self) {
@@ -87,19 +72,8 @@ impl NymApiTopologyProvider {
         }
 
         self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len();
-
-        // 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)
+        self.validator_client
+            .change_nym_api(self.nym_api_urls[self.currently_used_api].clone())
     }
 
     async fn get_current_compatible_topology(&mut self) -> Option<NymTopology> {
@@ -125,13 +99,8 @@ impl NymApiTopologyProvider {
                 .filter(|n| n.performance.round_to_integer() >= self.config.min_node_performance())
                 .collect::<Vec<_>>();
 
-            let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
-            NymTopology::new(
-                metadata.to_topology_metadata(),
-                epoch_rewarded_set,
-                Vec::new(),
-            )
-            .with_skimmed_nodes(&nodes_filtered)
+            NymTopology::new(metadata.to_topology_metadata(), 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
 
@@ -179,13 +148,8 @@ impl NymApiTopologyProvider {
                 }
             }
 
-            let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
-            NymTopology::new(
-                metadata.to_topology_metadata(),
-                epoch_rewarded_set,
-                Vec::new(),
-            )
-            .with_skimmed_nodes(&nodes)
+            NymTopology::new(metadata.to_topology_metadata(), rewarded_set, Vec::new())
+                .with_skimmed_nodes(&nodes)
         };
 
         if !topology.is_minimally_routable() {

common/client-core/src/error.rs

@@ -4,13 +4,10 @@
 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;
-use nym_validator_client::nyxd::error::NyxdError;
 use nym_validator_client::ValidatorClientError;
-use rand::distributions::WeightedError;
 use std::error::Error;
 use std::path::PathBuf;
 
@@ -57,7 +54,10 @@ pub enum ClientCoreError {
     ListOfNymApisIsEmpty,
 
     #[error("failed to resolve a query to nym API: {source}")]
-    NymApiQueryFailure { source: Box<NymAPIError> },
+    NymApiQueryFailure {
+        #[from]
+        source: NymAPIError,
+    },
 
     #[error(
         "the current network topology seem to be insufficient to route any packets through:\n\t{0}"
@@ -230,22 +230,7 @@ pub enum ClientCoreError {
     UnexpectedKeyUpgrade { gateway_id: String },
 
     #[error("failed to derive keys from master key")]
-    HkdfDerivationError,
-
-    #[error("missing url for constructing RPC client")]
-    RpcClientMissingUrl,
-
-    #[error("provided nym network details were malformed: {source}")]
-    InvalidNetworkDetails { source: NyxdError },
-
-    #[error("failed to construct RPC client: {source}")]
-    RpcClientCreationFailure { source: NyxdError },
-
-    #[error("failed to select valid gateway due to incomputable latency")]
-    GatewaySelectionFailure { source: WeightedError },
-
-    #[error("Could not access task registry, {0}")]
-    RegistryAccess(#[from] RegistryAccessError),
+    HkdfDerivationError {},
 }
 
 impl From<tungstenite::Error> for ClientCoreError {
@@ -256,14 +241,6 @@ 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,8 +7,7 @@ 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, NymApiClientExt};
-use nym_validator_client::nym_nodes::SkimmedNodesWithMetadata;
+use nym_validator_client::client::IdentityKeyRef;
 use nym_validator_client::UserAgent;
 use rand::{seq::SliceRandom, Rng};
 #[cfg(unix)]
@@ -84,48 +83,6 @@ 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 }
@@ -142,28 +99,16 @@ pub async fn gateways_for_init<R: Rng>(
     let nym_api = nym_apis
         .choose(rng)
         .ok_or(ClientCoreError::ListOfNymApisIsEmpty)?;
-
-    // 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))
-    })?;
+    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())
+    };
 
     tracing::debug!("Fetching list of gateways from: {nym_api}");
 
-    // Use our helper to handle pagination
-    let gateways = get_all_basic_entry_nodes_with_metadata(&client, true)
+    let gateways = client
+        .get_all_basic_entry_assigned_nodes_with_metadata()
         .await?
         .nodes;
     info!("nym api reports {} gateways", gateways.len());
@@ -203,7 +148,7 @@ async fn connect(endpoint: &str) -> Result<WsConn, ClientCoreError> {
     JSWebsocket::new(endpoint).map_err(|_| ClientCoreError::GatewayJsConnectionFailure)
 }
 
-async fn measure_latency<G>(gateway: &G) -> Result<GatewayWithLatency<'_, G>, ClientCoreError>
+async fn measure_latency<G>(gateway: &G) -> Result<GatewayWithLatency<G>, ClientCoreError>
 where
     G: ConnectableGateway,
 {
@@ -300,7 +245,7 @@ pub async fn choose_gateway_by_latency<R: Rng, G: ConnectableGateway + Clone>(
     let gateways_with_latency = gateways_with_latency.lock().await;
     let chosen = gateways_with_latency
         .choose_weighted(rng, |item| 1. / item.latency.as_secs_f32())
-        .map_err(|source| ClientCoreError::GatewaySelectionFailure { source })?;
+        .expect("invalid selection weight!");
 
     info!(
         "chose gateway {} with average latency of {:?}",

common/client-core/src/lib.rs

@@ -17,20 +17,16 @@ 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)
+pub(crate) fn spawn_future<F>(future: F)
 where
     F: Future<Output = ()> + 'static,
 {
     wasm_bindgen_futures::spawn_local(future);
 }
 
-#[deprecated(note = "use spawn_future from nym_task crate instead")]
 #[cfg(not(target_arch = "wasm32"))]
-#[track_caller]
-pub fn spawn_future<F>(future: F)
+pub(crate) fn spawn_future<F>(future: F)
 where
     F: Future + Send + 'static,
     F::Output: Send + 'static,

common/client-core/surb-storage/Cargo.toml

@@ -30,7 +30,6 @@ optional = true
 path = "../../../sqlx-pool-guard"
 
 [build-dependencies]
-anyhow = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 sqlx = { workspace = true, features = [
     "runtime-tokio-rustls",

common/client-core/surb-storage/build.rs

@@ -2,24 +2,23 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #[tokio::main]
-async fn main() -> anyhow::Result<()> {
+async fn main() {
     #[cfg(feature = "fs-surb-storage")]
     {
-        use anyhow::Context;
         use sqlx::{Connection, SqliteConnection};
         use std::env;
 
-        let out_dir = env::var("OUT_DIR")?;
+        let out_dir = env::var("OUT_DIR").unwrap();
         let database_path = format!("{out_dir}/fs-surbs-example.sqlite");
 
         let mut conn = SqliteConnection::connect(&format!("sqlite://{database_path}?mode=rwc"))
             .await
-            .context("Failed to create SQLx database connection")?;
+            .expect("Failed to create SQLx database connection");
 
         sqlx::migrate!("./fs_surbs_migrations")
             .run(&mut conn)
             .await
-            .context("Failed to perform SQLx migrations")?;
+            .expect("Failed to perform SQLx migrations");
 
         #[cfg(target_family = "unix")]
         println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path);
@@ -29,6 +28,4 @@ async fn main() -> anyhow::Result<()> {
         // not a valid windows path... but hey, it works...
         println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
     }
-
-    Ok(())
 }

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

@@ -40,7 +40,7 @@ where
     pub async fn flush_on_shutdown(
         mut self,
         mem_state: CombinedReplyStorage,
-        shutdown: nym_task::ShutdownToken,
+        mut shutdown: nym_task::TaskClient,
     ) {
         use tracing::{debug, error, info};
 
@@ -50,7 +50,7 @@ where
             return;
         }
 
-        shutdown.cancelled().await;
+        shutdown.recv().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;
+use nym_bandwidth_controller::{BandwidthController, BandwidthStatusMessage};
 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::ShutdownToken;
+use nym_task::TaskClient;
 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
-    shutdown_token: ShutdownToken,
+    task_client: TaskClient,
 }
 
 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>>,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self {
         GatewayClient {
             cfg,
@@ -141,7 +141,7 @@ impl<C, St> GatewayClient<C, St> {
             negotiated_protocol: None,
             #[cfg(unix)]
             connection_fd_callback,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -201,7 +201,7 @@ impl<C, St> GatewayClient<C, St> {
     #[cfg(not(target_arch = "wasm32"))]
     pub async fn establish_connection(&mut self) -> Result<(), GatewayClientError> {
         debug!(
-            "Attempting to establish connection to gateway at: {}",
+            "Attemting to establish connection to gateway at: {}",
             self.gateway_address
         );
         let (ws_stream, _) = connect_async(
@@ -293,7 +293,7 @@ impl<C, St> GatewayClient<C, St> {
 
         loop {
             tokio::select! {
-                _ = self.shutdown_token.cancelled() => {
+                _ = self.task_client.recv() => {
                     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.shutdown_token.clone(),
+                self.task_client.clone(),
             )
             .await
             .map_err(GatewayClientError::RegistrationFailure),
@@ -631,6 +631,9 @@ 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)),
@@ -1043,12 +1046,6 @@ impl<C, St> GatewayClient<C, St> {
     }
 
     // Note: this requires prior authentication
-    #[instrument(skip_all,
-        fields(
-            gateway = %self.gateway_identity,
-            gateway_address = %self.gateway_address
-        )
-    )]
     pub fn start_listening_for_mixnet_messages(&mut self) -> Result<(), GatewayClientError> {
         if !self.authenticated {
             return Err(GatewayClientError::NotAuthenticated);
@@ -1072,7 +1069,7 @@ impl<C, St> GatewayClient<C, St> {
                                 .expect("no shared key present even though we're authenticated!"),
                         ),
                         self.bandwidth.clone(),
-                        self.shutdown_token.clone(),
+                        self.task_client.clone(),
                     )
                 }
                 _ => unreachable!(),
@@ -1146,8 +1143,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 shutdown_token = ShutdownToken::default();
-        let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown_token.clone());
+        let task_client = TaskClient::dummy();
+        let packet_router = PacketRouter::new(ack_tx, mix_tx, task_client.clone());
 
         GatewayClient {
             cfg: GatewayClientConfig::default().with_disabled_credentials_mode(true),
@@ -1160,11 +1157,11 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
             connection: SocketState::NotConnected,
             packet_router,
             bandwidth_controller: None,
-            stats_reporter: ClientStatsSender::new(None, shutdown_token.clone()),
+            stats_reporter: ClientStatsSender::new(None, task_client.clone()),
             negotiated_protocol: None,
             #[cfg(unix)]
             connection_fd_callback,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -1173,7 +1170,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
         packet_router: PacketRouter,
         bandwidth_controller: Option<BandwidthController<C, St>>,
         stats_reporter: ClientStatsSender,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> GatewayClient<C, St> {
         // invariants that can't be broken
         // (unless somebody decided to expose some field that wasn't meant to be exposed)
@@ -1196,7 +1193,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
             negotiated_protocol: self.negotiated_protocol,
             #[cfg(unix)]
             connection_fd_callback: self.connection_fd_callback,
-            shutdown_token,
+            task_client,
         }
     }
 }

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::ShutdownToken;
+use nym_task::TaskClient;
 
 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: ShutdownToken,
+    shutdown: TaskClient,
 }
 
 impl PacketRouter {
     pub fn new(
         ack_sender: AcknowledgementSender,
         mixnet_message_sender: MixnetMessageSender,
-        shutdown: ShutdownToken,
+        shutdown: TaskClient,
     ) -> 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_cancelled() {
+            if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() {
                 // 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_cancelled() {
+            if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() {
                 // 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,6 +69,10 @@ 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::ShutdownToken;
+use nym_task::TaskClient;
 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>, shutdown_token: ShutdownToken) {
+    async fn run(mut self, mut split_stream: SplitStream<WsConn>, mut task_client: TaskClient) {
         let mut chunked_stream = (&mut split_stream).ready_chunks(8);
         let ret: Result<_, GatewayClientError> = loop {
             tokio::select! {
                 biased;
                 // received system-wide shutdown
-                _ = shutdown_token.cancelled() => {
+                _ = task_client.recv() => {
                     log::trace!("GatewayClient listener: Received shutdown");
                     log::debug!("GatewayClient listener: Exiting");
                     return;
@@ -118,7 +118,11 @@ impl PartiallyDelegatedRouter {
 
         let return_res = match ret {
             Err(err) => self.stream_return.send(Err(err)),
-            Ok(_) => self.stream_return.send(Ok(split_stream)),
+            Ok(_) => {
+                self.packet_router.disarm();
+                task_client.disarm();
+                self.stream_return.send(Ok(split_stream))
+            }
         };
 
         if return_res.is_err() {
@@ -262,8 +266,8 @@ impl PartiallyDelegatedRouter {
         Ok(plaintexts)
     }
 
-    fn spawn(self, split_stream: SplitStream<WsConn>, shutdown_token: ShutdownToken) {
-        let fut = async move { self.run(split_stream, shutdown_token).await };
+    fn spawn(self, split_stream: SplitStream<WsConn>, task_client: TaskClient) {
+        let fut = async move { self.run(split_stream, task_client).await };
 
         #[cfg(target_arch = "wasm32")]
         wasm_bindgen_futures::spawn_local(fut);
@@ -279,7 +283,7 @@ impl PartiallyDelegatedHandle {
         packet_router: PacketRouter,
         shared_key: Arc<SharedGatewayKey>,
         client_bandwidth: ClientBandwidth,
-        shutdown: ShutdownToken,
+        shutdown: TaskClient,
     ) -> Self {
         // when called for, it NEEDS TO yield back the stream so that we could merge it and
         // read control request responses.
@@ -333,7 +337,7 @@ impl PartiallyDelegatedHandle {
         // check if the split stream didn't error out
         let receive_res = stream_receiver
             .try_recv()
-            .map_err(|_| GatewayClientError::ConnectionAbruptlyClosed)?;
+            .expect("stream sender was somehow dropped without sending anything!");
 
         if let Some(res) = receive_res {
             let _res = res?;

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::{
-    DirectSigningReqwestRpcValidatorClient, QueryReqwestRpcValidatorClient, ReqwestRpcClient,
-    ValidatorClientError,
+    nym_api, DirectSigningReqwestRpcValidatorClient, QueryReqwestRpcValidatorClient,
+    ReqwestRpcClient, ValidatorClientError,
 };
 use nym_api_requests::ecash::models::{
     AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
@@ -20,9 +20,11 @@ use nym_api_requests::ecash::{
     PartialExpirationDateSignatureResponse, VerificationKeyResponse,
 };
 use nym_api_requests::models::{
-    ApiHealthResponse, GatewayCoreStatusResponse, HistoricalPerformanceResponse,
-    MixnodeCoreStatusResponse, NymNodeDescription,
+    ApiHealthResponse, GatewayBondAnnotated, GatewayCoreStatusResponse,
+    HistoricalPerformanceResponse, MixnodeCoreStatusResponse, MixnodeStatusResponse,
+    NymNodeDescription, RewardEstimationResponse, StakeSaturationResponse,
 };
+use nym_api_requests::models::{LegacyDescribedGateway, MixNodeBondAnnotated};
 use nym_api_requests::nym_nodes::{
     NodesByAddressesResponse, SemiSkimmedNodesWithMetadata, SkimmedNode, SkimmedNodesWithMetadata,
 };
@@ -151,7 +153,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_http_api_client::Client,
+    pub nym_api: nym_api::Client,
     // pub nym_api_client: NymApiClient,
     pub nyxd: NyxdClient<C, S>,
 }
@@ -212,7 +214,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_http_api_client::Client::new(config.api_url.clone(), None);
+        let nym_api_client = nym_api::Client::new(config.api_url.clone(), None);
 
         Client {
             nym_api: nym_api_client,
@@ -226,7 +228,7 @@ impl<C, S> Client<C, S> {
     where
         S: OfflineSigner,
     {
-        let nym_api_client = nym_http_api_client::Client::new(config.api_url.clone(), None);
+        let nym_api_client = nym_api::Client::new(config.api_url.clone(), None);
 
         Client {
             nym_api: nym_api_client,
@@ -247,6 +249,65 @@ 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,
@@ -324,25 +385,38 @@ 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_http_api_client::Client,
+    pub nym_api: nym_api::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_http_api_client::Client::new(api_url, Some(timeout));
+        let nym_api = nym_api::Client::new(api_url, Some(timeout));
 
         NymApiClient {
             use_bincode: true,
@@ -357,10 +431,10 @@ impl NymApiClient {
     }
 
     pub fn new_with_user_agent(api_url: Url, user_agent: impl Into<UserAgent>) -> Self {
-        let nym_api = nym_http_api_client::Client::builder(api_url)
+        let nym_api = nym_api::Client::builder::<_, ValidatorClientError>(api_url)
             .expect("invalid api url")
             .with_user_agent(user_agent.into())
-            .build()
+            .build::<ValidatorClientError>()
             .expect("failed to build nym api client");
 
         NymApiClient {
@@ -497,6 +571,37 @@ 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> {
@@ -563,6 +668,30 @@ 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,
@@ -590,11 +719,10 @@ impl NymApiClient {
     pub async fn partial_expiration_date_signatures(
         &self,
         expiration_date: Option<Date>,
-        epoch_id: Option<EpochId>,
     ) -> Result<PartialExpirationDateSignatureResponse, ValidatorClientError> {
         Ok(self
             .nym_api
-            .partial_expiration_date_signatures(expiration_date, epoch_id)
+            .partial_expiration_date_signatures(expiration_date)
             .await?)
     }
 
@@ -611,11 +739,10 @@ impl NymApiClient {
     pub async fn global_expiration_date_signatures(
         &self,
         expiration_date: Option<Date>,
-        epoch_id: Option<EpochId>,
     ) -> Result<AggregatedExpirationDateSignatureResponse, ValidatorClientError> {
         Ok(self
             .nym_api
-            .global_expiration_date_signatures(expiration_date, epoch_id)
+            .global_expiration_date_signatures(expiration_date)
             .await?)
     }
 

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

@@ -3,6 +3,7 @@
 
 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;
@@ -14,7 +15,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: nym_http_api_client::Client,
+    pub api_client: NymApiClient,
     pub verification_key: VerificationKeyAuth,
     pub node_id: NodeIndex,
     pub cosmos_address: cosmrs::AccountId,
@@ -24,15 +25,10 @@ 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
-                .base_urls()
-                .iter()
-                .map(|url| url.to_string())
-                .collect::<Vec<String>>()
-                .join(", ")
+            self.api_client.api_url()
         )
     }
 }
@@ -64,9 +60,6 @@ pub enum EcashApiError {
         source: CompactEcashError,
     },
 
-    #[error("failed to create API client: {0}")]
-    ClientError(String),
-
     #[error("the provided account address is malformed: {source}")]
     MalformedAccountAddress {
         #[from]
@@ -96,13 +89,8 @@ 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,
+            api_client: NymApiClient::new(url_address),
             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,8 +1,7 @@
-use crate::nym_api::NymApiClientExt;
 use crate::nyxd::contract_traits::MixnetQueryClient;
 use crate::nyxd::error::NyxdError;
 use crate::nyxd::Config as ClientConfig;
-use crate::{QueryHttpRpcNyxdClient, ValidatorClientError};
+use crate::{NymApiClient, QueryHttpRpcNyxdClient, ValidatorClientError};
 use colored::Colorize;
 use core::fmt;
 use itertools::Itertools;
@@ -88,17 +87,8 @@ fn setup_connection_tests<H: BuildHasher + 'static>(
         }
     });
 
-    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
-            }
-        }
+    let api_connection_test_clients = api_urls.map(|(network, url)| {
+        ClientForConnectionTest::Api(network, url.clone(), NymApiClient::new(url))
     });
 
     nyxd_connection_test_clients.chain(api_connection_test_clients)
@@ -170,7 +160,7 @@ async fn test_nyxd_connection(
 async fn test_nym_api_connection(
     network: NymNetworkDetails,
     url: &Url,
-    client: &nym_http_api_client::Client,
+    client: &NymApiClient,
 ) -> ConnectionResult {
     let result = match timeout(
         Duration::from_secs(CONNECTION_TEST_TIMEOUT_SEC),
@@ -196,7 +186,7 @@ async fn test_nym_api_connection(
 
 enum ClientForConnectionTest {
     Nyxd(NymNetworkDetails, Url, Box<QueryHttpRpcNyxdClient>),
-    Api(NymNetworkDetails, Url, nym_http_api_client::Client),
+    Api(NymNetworkDetails, Url, NymApiClient),
 }
 
 impl ClientForConnectionTest {

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

@@ -9,7 +9,8 @@ use thiserror::Error;
 pub enum ValidatorClientError {
     #[error("nym api request failed: {source}")]
     NymAPIError {
-        source: Box<nym_api::error::NymAPIError>,
+        #[from]
+        source: nym_api::error::NymAPIError,
     },
 
     #[error("Tendermint RPC request failure: {0}")]
@@ -27,11 +28,3 @@ 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,6 +14,7 @@ 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,6 +1,7 @@
 // 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;
+pub type NymAPIError = HttpClientError<RequestError>;

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

@@ -3,21 +3,19 @@
 
 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,
-    BatchRedeemTicketsBody, EcashBatchTicketRedemptionResponse, EcashSignerStatusResponse,
-    EcashTicketVerificationResponse, IssuedTicketbooksChallengeCommitmentRequest,
-    IssuedTicketbooksChallengeCommitmentResponse, IssuedTicketbooksDataRequest,
-    IssuedTicketbooksDataResponse, IssuedTicketbooksForCountResponse, IssuedTicketbooksForResponse,
-    VerifyEcashTicketBody,
+    BatchRedeemTicketsBody, EcashBatchTicketRedemptionResponse, EcashTicketVerificationResponse,
+    IssuedTicketbooksChallengeCommitmentRequest, IssuedTicketbooksChallengeCommitmentResponse,
+    IssuedTicketbooksDataRequest, IssuedTicketbooksDataResponse, IssuedTicketbooksForCountResponse,
+    IssuedTicketbooksForResponse, VerifyEcashTicketBody,
 };
 use nym_api_requests::ecash::VerificationKeyResponse;
 use nym_api_requests::models::{
-    AnnotationResponse, ApiHealthResponse, BinaryBuildInformationOwned, ChainBlocksStatusResponse,
-    ChainStatusResponse, KeyRotationInfoResponse, NodePerformanceResponse, NodeRefreshBody,
-    NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse, SignerInformationResponse,
+    AnnotationResponse, ApiHealthResponse, BinaryBuildInformationOwned, ChainStatusResponse,
+    KeyRotationInfoResponse, LegacyDescribedMixNode, NodePerformanceResponse, NodeRefreshBody,
+    NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
 };
 use nym_api_requests::nym_nodes::{
     NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponseV1,
@@ -31,22 +29,26 @@ pub use nym_api_requests::{
         VerifyEcashCredentialBody,
     },
     models::{
-        GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse,
-        MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse,
-        MixnodeUptimeHistoryResponse, StakeSaturationResponse, UptimeResponse,
+        ComputeRewardEstParam, GatewayBondAnnotated, GatewayCoreStatusResponse,
+        GatewayStatusReportResponse, GatewayUptimeHistoryResponse, LegacyDescribedGateway,
+        MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse,
+        MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse,
+        StakeSaturationResponse, UptimeResponse,
     },
-    nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SemiSkimmedNodesWithMetadata, SkimmedNode},
+    nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SkimmedNode},
     NymNetworkDetailsResponse,
 };
+use nym_contracts_common::IdentityKey;
 use nym_http_api_client::{ApiClient, NO_PARAMS};
-use nym_mixnet_contract_common::{IdentityKeyRef, NodeId, NymNodeDetails};
+use nym_mixnet_contract_common::mixnode::MixNodeDetails;
+use nym_mixnet_contract_common::{GatewayBond, 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;
@@ -58,9 +60,6 @@ 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(
             &[
@@ -86,6 +85,104 @@ 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,
@@ -142,156 +239,6 @@ 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,
@@ -319,25 +266,6 @@ 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> {
@@ -747,6 +675,42 @@ 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(
@@ -823,6 +787,24 @@ 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(
@@ -890,6 +872,104 @@ 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,
@@ -938,6 +1018,34 @@ 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,
@@ -993,9 +1101,8 @@ pub trait NymApiClientExt: ApiClient {
     async fn partial_expiration_date_signatures(
         &self,
         expiration_date: Option<Date>,
-        epoch_id: Option<EpochId>,
     ) -> Result<PartialExpirationDateSignatureResponse, NymAPIError> {
-        let mut params = match expiration_date {
+        let params = match expiration_date {
             None => Vec::new(),
             Some(exp) => vec![(
                 ecash::EXPIRATION_DATE_PARAM,
@@ -1003,10 +1110,6 @@ pub trait NymApiClientExt: ApiClient {
             )],
         };
 
-        if let Some(epoch_id) = epoch_id {
-            params.push((ecash::EPOCH_ID_PARAM, epoch_id.to_string()));
-        }
-
         self.get_json(
             &[
                 routes::V1_API_VERSION,
@@ -1043,9 +1146,8 @@ pub trait NymApiClientExt: ApiClient {
     async fn global_expiration_date_signatures(
         &self,
         expiration_date: Option<Date>,
-        epoch_id: Option<EpochId>,
     ) -> Result<AggregatedExpirationDateSignatureResponse, NymAPIError> {
-        let mut params = match expiration_date {
+        let params = match expiration_date {
             None => Vec::new(),
             Some(exp) => vec![(
                 ecash::EXPIRATION_DATE_PARAM,
@@ -1053,10 +1155,6 @@ pub trait NymApiClientExt: ApiClient {
             )],
         };
 
-        if let Some(epoch_id) = epoch_id {
-            params.push((ecash::EPOCH_ID_PARAM, epoch_id.to_string()));
-        }
-
         self.get_json(
             &[
                 routes::V1_API_VERSION,
@@ -1233,22 +1331,6 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
-    async fn get_chain_blocks_status(&self) -> Result<ChainBlocksStatusResponse, NymAPIError> {
-        self.get_json("/v1/network/chain-blocks-status", NO_PARAMS)
-            .await
-    }
-
-    #[instrument(level = "debug", skip(self))]
-    async fn get_signer_status(&self) -> Result<EcashSignerStatusResponse, NymAPIError> {
-        self.get_json("/v1/ecash/signer-status", NO_PARAMS).await
-    }
-
-    #[instrument(level = "debug", skip(self))]
-    async fn get_signer_information(&self) -> Result<SignerInformationResponse, NymAPIError> {
-        self.get_json("/v1/api-status/signer-information", NO_PARAMS)
-            .await
-    }
-
     #[instrument(level = "debug", skip(self))]
     async fn get_key_rotation_info(&self) -> Result<KeyRotationInfoResponse, NymAPIError> {
         self.get_json(
@@ -1261,49 +1343,8 @@ 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 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());
-    }
-}
+impl NymApiClientExt for Client {}

common/client-libs/validator-client/src/nyxd/contract_traits/dkg_query_client.rs

@@ -8,11 +8,11 @@ use crate::nyxd::CosmWasmClient;
 use async_trait::async_trait;
 use cosmrs::AccountId;
 use cosmwasm_std::Addr;
-use nym_coconut_dkg_common::dealer::RegisteredDealerDetails;
 use nym_coconut_dkg_common::types::{ChunkIndex, NodeIndex, StateAdvanceResponse};
 use serde::Deserialize;
 use tracing::trace;
 
+use nym_coconut_dkg_common::dealer::RegisteredDealerDetails;
 pub use nym_coconut_dkg_common::{
     dealer::{DealerDetailsResponse, PagedDealerIndexResponse, PagedDealerResponse},
     dealing::{
@@ -21,9 +21,7 @@ pub use nym_coconut_dkg_common::{
     },
     msg::QueryMsg as DkgQueryMsg,
     types::{DealerDetails, DealingIndex, Epoch, EpochId, EpochState, State},
-    verification_key::{
-        ContractVKShare, PagedVKSharesResponse, VerificationKeyShare, VkShareResponse,
-    },
+    verification_key::{ContractVKShare, PagedVKSharesResponse, VkShareResponse},
 };
 
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]

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};
+use cosmrs::{AccountId, Coin as CosmosCoin, Tx};
 use prost::Message;
 use serde::{Deserialize, Serialize};
 
@@ -556,12 +556,23 @@ pub trait CosmWasmClient: TendermintRpcClient {
         Ok(serde_json::from_slice(&res.data)?)
     }
 
-    async fn query_simulate(&self, tx_bytes: Vec<u8>) -> Result<SimulateResponse, NyxdError> {
+    // 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> {
         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,14 +81,17 @@ 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
 
-        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
+        // 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
     }
 
     async fn upload(

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

@@ -139,22 +139,12 @@ impl NyxdClient<HttpClient> {
         })
     }
 
-    pub fn connect_with_network_details<U>(
-        endpoint: U,
-        network_details: NymNetworkDetails,
-    ) -> Result<QueryHttpRpcNyxdClient, NyxdError>
-    where
-        U: TryInto<HttpClientUrl, Error = TendermintRpcError>,
-    {
-        let config = Config::try_from_nym_network_details(&network_details)?;
-        Self::connect(config, endpoint)
-    }
-
     pub fn connect_to_default_env<U>(endpoint: U) -> Result<QueryHttpRpcNyxdClient, NyxdError>
     where
         U: TryInto<HttpClientUrl, Error = TendermintRpcError>,
     {
-        Self::connect_with_network_details(endpoint, NymNetworkDetails::new_from_env())
+        let config = Config::try_from_nym_network_details(&NymNetworkDetails::new_from_env())?;
+        Self::connect(config, endpoint)
     }
 }
 

common/commands/Cargo.toml

@@ -38,7 +38,6 @@ 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/commands/src/ecash/generate_ticket.rs

@@ -86,7 +86,7 @@ pub async fn execute(args: Args) -> anyhow::Result<()> {
             anyhow!("ticketbook got incorrectly imported - the master verification key is missing")
         })?;
     let expiration_signatures = persistent_storage
-        .get_expiration_date_signatures(expiration_date, epoch_id)
+        .get_expiration_date_signatures(expiration_date)
         .await?
         .ok_or_else(|| {
             anyhow!(

common/commands/src/ecash/issue_ticket_book.rs

@@ -120,7 +120,7 @@ async fn issue_to_file(args: Args, client: SigningClient) -> anyhow::Result<()>
 
     if args.include_expiration_date_signatures {
         let signatures = credentials_store
-            .get_expiration_date_signatures(expiration_date, epoch_id)
+            .get_expiration_date_signatures(expiration_date)
             .await?
             .ok_or(anyhow!("missing expiration date signatures!"))?;
 

common/commands/src/internal/mod.rs

@@ -4,7 +4,6 @@
 use clap::{Args, Subcommand};
 
 pub mod ecash;
-pub mod nyx;
 
 #[derive(Debug, Args)]
 #[clap(args_conflicts_with_subcommands = true, subcommand_required = true)]
@@ -17,6 +16,4 @@ pub struct Internal {
 pub enum InternalCommands {
     /// Ecash related internal commands
     Ecash(ecash::InternalEcash),
-
-    Nyx(nyx::InternalNyx),
 }

common/commands/src/internal/nyx/force_advance_epoch.rs

@@ -1,116 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use crate::context::SigningClient;
-use anyhow::bail;
-use clap::Parser;
-use nym_mixnet_contract_common::nym_node::Role;
-use nym_mixnet_contract_common::reward_params::NodeRewardingParameters;
-use nym_mixnet_contract_common::{
-    EpochRewardedSet, EpochState, NodeId, RewardingParams, RoleAssignment,
-};
-use nym_validator_client::nyxd::contract_traits::mixnet_query_client::MixnetQueryClientExt;
-use nym_validator_client::nyxd::contract_traits::{MixnetQueryClient, MixnetSigningClient};
-use rand::prelude::*;
-use rand::thread_rng;
-
-#[derive(Debug, Parser)]
-pub struct Args {}
-
-fn choose_new_nodes(
-    params: &RewardingParams,
-    rewarded_set: &EpochRewardedSet,
-    role: Role,
-) -> Vec<NodeId> {
-    let mut rng = thread_rng();
-
-    match role {
-        Role::EntryGateway => rewarded_set
-            .assignment
-            .entry_gateways
-            .choose_multiple(&mut rng, params.rewarded_set.entry_gateways as usize)
-            .copied()
-            .collect(),
-        Role::Layer1 => rewarded_set
-            .assignment
-            .layer1
-            .choose_multiple(&mut rng, params.rewarded_set.mixnodes as usize / 3)
-            .copied()
-            .collect(),
-        Role::Layer2 => rewarded_set
-            .assignment
-            .layer2
-            .choose_multiple(&mut rng, params.rewarded_set.mixnodes as usize / 3)
-            .copied()
-            .collect(),
-        Role::Layer3 => rewarded_set
-            .assignment
-            .layer3
-            .choose_multiple(&mut rng, params.rewarded_set.mixnodes as usize / 3)
-            .copied()
-            .collect(),
-        Role::ExitGateway => rewarded_set
-            .assignment
-            .exit_gateways
-            .choose_multiple(&mut rng, params.rewarded_set.exit_gateways as usize)
-            .copied()
-            .collect(),
-        Role::Standby => rewarded_set
-            .assignment
-            .standby
-            .choose_multiple(&mut rng, params.rewarded_set.standby as usize)
-            .copied()
-            .collect(),
-    }
-}
-
-pub async fn force_advance_epoch(_: Args, client: SigningClient) -> anyhow::Result<()> {
-    let current_epoch = client.get_current_interval_details().await?;
-    let epoch_status = client.get_current_epoch_status().await?;
-    if epoch_status.being_advanced_by.as_str() != client.address().to_string() {
-        bail!(
-            "this client is not authorised to perform any epoch operations. we need {}",
-            client.address()
-        )
-    }
-
-    let rewarding_params = client.get_rewarding_parameters().await?;
-    let current_rewarded_set = client.get_rewarded_set().await?;
-
-    if !current_epoch.is_current_epoch_over {
-        println!("the current epoch is not over yet - there's nothing to do")
-    }
-
-    // is this most efficient? no. but it's simple
-    loop {
-        let epoch_status = client.get_current_epoch_status().await?;
-
-        match epoch_status.state {
-            EpochState::InProgress => break,
-            EpochState::Rewarding { final_node_id, .. } => {
-                println!("rewarding {final_node_id} with big fat 0...");
-                client
-                    .reward_node(
-                        final_node_id,
-                        NodeRewardingParameters::new(Default::default(), Default::default()),
-                        None,
-                    )
-                    .await?;
-            }
-            EpochState::ReconcilingEvents => {
-                println!("trying to reconcile events...");
-                client.reconcile_epoch_events(None, None).await?;
-            }
-            EpochState::RoleAssignment { next } => {
-                let nodes = choose_new_nodes(&rewarding_params, &current_rewarded_set, next);
-                println!("assigning {nodes:?} as {next}");
-
-                client
-                    .assign_roles(RoleAssignment { role: next, nodes }, None)
-                    .await?;
-            }
-        }
-    }
-
-    Ok(())
-}

common/commands/src/internal/nyx/mod.rs

@@ -1,19 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use clap::{Args, Subcommand};
-
-pub mod force_advance_epoch;
-
-#[derive(Debug, Args)]
-#[clap(args_conflicts_with_subcommands = true, subcommand_required = true)]
-pub struct InternalNyx {
-    #[clap(subcommand)]
-    pub command: InternalNyxCommands,
-}
-
-#[derive(Debug, Subcommand)]
-pub enum InternalNyxCommands {
-    /// Attempt to force advance the current epoch
-    ForceAdvanceEpoch(force_advance_epoch::Args),
-}

common/commands/src/validator/mixnet/delegators/delegate_to_multiple_mixnodes.rs

@@ -241,7 +241,7 @@ pub async fn delegate_to_multiple_mixnodes(args: Args, client: SigningClient) {
                 let node_id = row.node_id.clone().parse::<u32>().unwrap();
                 let coins: Vec<Coin> = vec![];
                 undelegation_msgs.push((ExecuteMsg::Undelegate { node_id }, coins));
-                undelegation_table.add_row(std::slice::from_ref(&row.node_id));
+                undelegation_table.add_row(&[row.node_id.clone()]);
 
                 if row.amount.amount > 0 {
                     delegation_msgs

common/cosmwasm-smart-contracts/contracts-common-testing/src/tester/mod.rs

@@ -188,7 +188,7 @@ impl<C> ContractTesterBuilder<C> {
         *self.app.api()
     }
 
-    pub fn querier(&self) -> QuerierWrapper<'_> {
+    pub fn querier(&self) -> QuerierWrapper {
         self.app.wrap()
     }
 }

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 = "entry_gateway" | "layer1" | "layer2" | "layer3" | "exit_gateway" | "standby";
+export type Role = "EntryGateway" | "Layer1" | "Layer2" | "Layer3" | "ExitGateway" | "Standby";