changes due to crate moving

Previously, when rotating API endpoints, only a single URL was provided to the
HTTP client, defeating the purpose of having multiple URLs for resilience.

Changes:
- NymApiTopologyProvider now provides all URLs in rotated order when switching endpoints
- NymApisClient similarly provides all URLs starting from the working endpoint
- Added clarifying comments for broadcast/exhaustive query methods where single URLs are intentionally used
- This enables the HTTP client's built-in failover mechanism while maintaining endpoint rotation behavior

The fix ensures that if the primary endpoint fails, the client can automatically
failover to alternative endpoints without manual intervention, improving overall
network resilience.

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

.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

@@ -81,21 +81,12 @@ jobs:
           command: fmt
           args: --all -- --check
 
-      - name: Clippy (macos)
-        if: contains(matrix.os, 'mac')
-        uses: actions-rs/cargo@v1
-        with:
-          command: clippy
-          args: --workspace --all-targets --exclude nym-gateway-probe -- -D warnings
-
-      - name: Clippy (non-macos)
-        if: contains(matrix.os, 'linux') || contains(matrix.os, 'windows')
+      - name: Clippy
         uses: actions-rs/cargo@v1
         with:
           command: clippy
           args: --workspace --all-targets -- -D warnings
 
-
       - name: Build all binaries
         uses: actions-rs/cargo@v1
         with:

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

@@ -10,13 +10,13 @@ env:
 
 jobs:
   check-if-tag-exists:
-    runs-on: arc-linux-latest-dind
+    runs-on: arc-ubuntu-22.04-dind
     steps:
       - name: Checkout repo
         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-contracts-upload-binaries.yml

@@ -11,7 +11,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ arc-linux-latest-dind ]
+        platform: [ arc-ubuntu-22.04 ]
 
     runs-on: ${{ matrix.platform }}
     env:
@@ -28,11 +28,18 @@ jobs:
           mkdir -p $OUTPUT_DIR
           echo $OUTPUT_DIR
 
-      - name: Build contracts
-        run: make optimize-contracts
+      - name: Install Rust stable
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          target: wasm32-unknown-unknown
+          override: true
 
-      - name: Check optimized contracts
-        run: make docker-check-contracts
+      - name: Install cosmwasm-check
+        run: cargo install cosmwasm-check
+
+      - name: Build release contracts
+        run: make publish-contracts
 
       - name: Prepare build output
         shell: bash

.github/workflows/ci-contracts.yml

@@ -17,7 +17,7 @@ jobs:
   build:
     # since it's going to be compiled into wasm, there's absolutely
     # no point in running CI on different OS-es
-    runs-on: arc-linux-latest
+    runs-on: ubuntu-22.04
     env:
       CARGO_TERM_COLOR: always
       RUSTUP_PERMIT_COPY_RENAME: 1
@@ -54,7 +54,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --lib --manifest-path contracts/Cargo.toml --all-features
+          args: --lib --manifest-path contracts/Cargo.toml
 
       - name: Check formatting
         uses: actions-rs/cargo@v1

.github/workflows/ci-docs.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   build:
-    runs-on: arc-linux-latest
+    runs-on: arc-ubuntu-22.04
     env:
       RUSTUP_PERMIT_COPY_RENAME: 1
     defaults:

.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

@@ -4,14 +4,14 @@ on:
   workflow_dispatch:
   pull_request:
     paths:
-      - "wasm/**"
-      - "clients/client-core/**"
-      - "common/**"
-      - ".github/workflows/ci-sdk-wasm.yml"
+      - 'wasm/**'
+      - 'clients/client-core/**'
+      - 'common/**'
+      - '.github/workflows/ci-sdk-wasm.yml'
 
 jobs:
   wasm:
-    runs-on: arc-linux-latest
+    runs-on: arc-ubuntu-22.04
     env:
       CARGO_TERM_COLOR: always
       RUSTUP_PERMIT_COPY_RENAME: 1
@@ -33,7 +33,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v5
         with:
-          go-version: "1.24.6"
+          go-version: "1.23.7"
 
       - name: Install wasm-pack
         run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
@@ -41,7 +41,7 @@ jobs:
       - name: Install wasm-opt
         uses: ./.github/actions/install-wasm-opt
         with:
-          version: "116"
+          version: '116'
 
       - name: Install wasm-bindgen-cli
         run: cargo install wasm-bindgen-cli

.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

@@ -21,7 +21,7 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - os: arc-linux-latest
+          - os: arc-ubuntu-22.04
             target: x86_64-unknown-linux-gnu
     runs-on:  ${{ matrix.os }}
 
@@ -56,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

.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

@@ -8,7 +8,7 @@ env:
 
 jobs:
   build-container:
-    runs-on: arc-linux-latest-dind
+    runs-on: arc-ubuntu-22.04-dind
     steps:
       - name: Login to Harbor
         uses: docker/login-action@v3
@@ -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

@@ -20,7 +20,7 @@ env:
 
 jobs:
   build-container:
-    runs-on: arc-linux-latest-dind
+    runs-on: arc-ubuntu-22.04-dind
     steps:
       - name: Login to Harbor
         uses: docker/login-action@v3

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

@@ -14,7 +14,7 @@ env:
 
 jobs:
   build-container:
-    runs-on: arc-linux-latest-dind
+    runs-on: arc-ubuntu-22.04-dind
     steps:
       - name: Login to Harbor
         uses: docker/login-action@v3
@@ -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

@@ -8,7 +8,7 @@ env:
 
 jobs:
   build-container:
-    runs-on: arc-linux-latest-dind
+    runs-on: arc-ubuntu-22.04-dind
     steps:
       - name: Login to Harbor
         uses: docker/login-action@v3
@@ -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,170 +4,6 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ## [Unreleased]
 
-## [2025.18-jarlsberg] (2025-10-14)
-
-- ns-api: add descriptions to dVPN gateway responses ([#6102])
-- NS API: use new probe download filesize and milliseconds field ([#6097])
-- ns-api: use download files size from probes instead of parsing filenames ([#6095])
-- ns-api: add new fields for probe output for query_metadata and download file size and duration in ms ([#6091])
-- Bugfix/bloomfilters purge ([#6089])
-- Hotfix: Update API source in node ping tester script ([#6082])
-- Get wireguard keypair as arg instead of reading it from disk ([#6078])
-- Feature: Ping probe all nodes /described nodes from a server ([#6074])
-- Node Status API: add bridge information to dVPN endpoint ([#6069])
-- frontdoor typo fix ([#6067])
-- Feature: Node rewards tracker ([#6064])
-- [chore] Clippy fix ([#6060])
-- Registration Client ([#6059])
-- Bugfix: Nym node CLI download nym-node exception ([#6058])
-- Feature: Nym node html landing page ([#6053])
-- feat: DKG contract method for updating announce address ([#6050])
-- feat: NS ticket faucet ([#6047])
-- Bridge proto client params in Self-Described ([#6035])
-- Node Status API: remove sqlite support ([#6004])
-- Benny/ci contract fix ([#5962])
-
-[#6102]: https://github.com/nymtech/nym/pull/6102
-[#6097]: https://github.com/nymtech/nym/pull/6097
-[#6095]: https://github.com/nymtech/nym/pull/6095
-[#6091]: https://github.com/nymtech/nym/pull/6091
-[#6089]: https://github.com/nymtech/nym/pull/6089
-[#6082]: https://github.com/nymtech/nym/pull/6082
-[#6078]: https://github.com/nymtech/nym/pull/6078
-[#6074]: https://github.com/nymtech/nym/pull/6074
-[#6069]: https://github.com/nymtech/nym/pull/6069
-[#6067]: https://github.com/nymtech/nym/pull/6067
-[#6064]: https://github.com/nymtech/nym/pull/6064
-[#6060]: https://github.com/nymtech/nym/pull/6060
-[#6059]: https://github.com/nymtech/nym/pull/6059
-[#6058]: https://github.com/nymtech/nym/pull/6058
-[#6053]: https://github.com/nymtech/nym/pull/6053
-[#6050]: https://github.com/nymtech/nym/pull/6050
-[#6047]: https://github.com/nymtech/nym/pull/6047
-[#6035]: https://github.com/nymtech/nym/pull/6035
-[#6004]: https://github.com/nymtech/nym/pull/6004
-[#5962]: https://github.com/nymtech/nym/pull/5962
-
-## [2025.17-isabirra] (2025-09-29)
-
-- Bugfix | Fix the registration handshake ([#6062])
-- Convenience for ShutdownTracker ([#6038])
-- chore: made http-api-client-macro doctest compile ([#6037])
-- feat: refresh mixnet contract on epoch progression ([#6023])
-- chore: remove legacy nodes from nym api [and kinda-ish from node status api] ([#6021])
-- Feature/credential proxy crate ([#6018])
-- Moving clients crate from vpn-client repo to here ([#6015])
-- Feature/cancellation migration ([#6014])
-- Use default value for the ports until api is deployed ([#6007])
-- bugfix: return from MixTrafficController if client request channel has closed ([#6002])
-- Revert "Create an axum_test client for more integrated unit testing (… ([#5999])
-- chore: upgraded syn to 2.0 and removed nym-execute ([#5998])
-- feat: use `ShutdownToken` (`CancellationToken` inside) for nym-api ([#5997])
-- bugfix: Recipient deserialisation for deserialisers missing bytes specialisation ([#5991])
-- chore: use updated version of simulate endpoint ([#5988])
-- chore: purge temp databases on build ([#5984])
-- Bump sha.js from 2.4.11 to 2.4.12 ([#5983])
-- Feature: Delegation program stake checker and adjuster ([#5980])
-- build(deps): bump actions/setup-java from 4 to 5 ([#5975])
-- Domain fronting integration ([#5974])
-- chore: internal hidden command to force advance nyx epoch ([#5964])
-- Create an axum_test client for more integrated unit testing ([#5956])
-- feat: shared library for attempting to retrieve update mode attestation ([#5954])
-- Bump slab from 0.4.10 to 0.4.11 ([#5952])
-- build(deps): bump actions/first-interaction from 1 to 3 ([#5950])
-- fix: use WASM compatible time API in client ([#5948])
-- feat: credential proxy deposit pool ([#5945])
-- build(deps): bump actions/download-artifact from 4 to 5 ([#5939])
-- feat: nym signers monitor ([#5933])
-- Bump console from 0.15.11 to 0.16.0 ([#5931])
-- Bump mock_instant from 0.5.3 to 0.6.0 ([#5930])
-- Bump tokio from 1.46.1 to 1.47.1 ([#5929])
-- Bump defguard_wireguard_rs from v0.4.7 to v0.7.5 ([#5928])
-- Bump indicatif from 0.17.11 to 0.18.0 ([#5924])
-- Feature: Nym node autorun CLI ([#5916])
-- build(deps): bump mikefarah/yq from 4.45.4 to 4.47.1 ([#5911])
-- build(deps): bump pbkdf2 from 3.1.2 to 3.1.3 ([#5869])
-
-[#6062]: https://github.com/nymtech/nym/pull/6062
-[#6038]: https://github.com/nymtech/nym/pull/6038
-[#6037]: https://github.com/nymtech/nym/pull/6037
-[#6023]: https://github.com/nymtech/nym/pull/6023
-[#6021]: https://github.com/nymtech/nym/pull/6021
-[#6018]: https://github.com/nymtech/nym/pull/6018
-[#6015]: https://github.com/nymtech/nym/pull/6015
-[#6014]: https://github.com/nymtech/nym/pull/6014
-[#6007]: https://github.com/nymtech/nym/pull/6007
-[#6002]: https://github.com/nymtech/nym/pull/6002
-[#5999]: https://github.com/nymtech/nym/pull/5999
-[#5998]: https://github.com/nymtech/nym/pull/5998
-[#5997]: https://github.com/nymtech/nym/pull/5997
-[#5991]: https://github.com/nymtech/nym/pull/5991
-[#5988]: https://github.com/nymtech/nym/pull/5988
-[#5984]: https://github.com/nymtech/nym/pull/5984
-[#5983]: https://github.com/nymtech/nym/pull/5983
-[#5980]: https://github.com/nymtech/nym/pull/5980
-[#5975]: https://github.com/nymtech/nym/pull/5975
-[#5974]: https://github.com/nymtech/nym/pull/5974
-[#5964]: https://github.com/nymtech/nym/pull/5964
-[#5956]: https://github.com/nymtech/nym/pull/5956
-[#5954]: https://github.com/nymtech/nym/pull/5954
-[#5952]: https://github.com/nymtech/nym/pull/5952
-[#5950]: https://github.com/nymtech/nym/pull/5950
-[#5948]: https://github.com/nymtech/nym/pull/5948
-[#5945]: https://github.com/nymtech/nym/pull/5945
-[#5939]: https://github.com/nymtech/nym/pull/5939
-[#5933]: https://github.com/nymtech/nym/pull/5933
-[#5931]: https://github.com/nymtech/nym/pull/5931
-[#5930]: https://github.com/nymtech/nym/pull/5930
-[#5929]: https://github.com/nymtech/nym/pull/5929
-[#5928]: https://github.com/nymtech/nym/pull/5928
-[#5924]: https://github.com/nymtech/nym/pull/5924
-[#5916]: https://github.com/nymtech/nym/pull/5916
-[#5911]: https://github.com/nymtech/nym/pull/5911
-[#5869]: https://github.com/nymtech/nym/pull/5869
-
-## [2025.16-halloumi] (2025-09-16)
-
-- Backport metadata endpoint ([#6010])
-- bugfix: make sure tables are removed in correct order to not trigger FK constraint issue ([#5987])
-- chore: move authenticator into gateway crate ([#5982])
-- Fix the ns api ci workflow ([#5981])
-- Remove freshness check on testrun submit ([#5977])
-- Update sysinfo to the latest ([#5976])
-- bugfix: manually calculate per node work on rewarded set changes ([#5972])
-- fixing the ci for ns agent ([#5965])
-- Feature/testing utils ([#5963])
-- bugfix: fix ci-build for linux (and use updated runner) ([#5958])
-- chore: updated refs to cheddar rev of nym repo ([#5955])
-- http api client adjustment ([#5953])
-- chore: fix rust 1.89 clippy issues ([#5944])
-- Wireguard metadata client library ([#5943])
-- chore: remove unused import ([#5942])
-- feat: introduce additional checks when attempting to send to bounded channels ([#5941])
-- Move credential verifier in peer controller ([#5938])
-- change PK/FK on expiration date signatures tables ([#5934])
-- Wireguard private metadata ([#5915])
-
-[#6010]: https://github.com/nymtech/nym/pull/6010
-[#5987]: https://github.com/nymtech/nym/pull/5987
-[#5982]: https://github.com/nymtech/nym/pull/5982
-[#5981]: https://github.com/nymtech/nym/pull/5981
-[#5977]: https://github.com/nymtech/nym/pull/5977
-[#5976]: https://github.com/nymtech/nym/pull/5976
-[#5972]: https://github.com/nymtech/nym/pull/5972
-[#5965]: https://github.com/nymtech/nym/pull/5965
-[#5963]: https://github.com/nymtech/nym/pull/5963
-[#5958]: https://github.com/nymtech/nym/pull/5958
-[#5955]: https://github.com/nymtech/nym/pull/5955
-[#5953]: https://github.com/nymtech/nym/pull/5953
-[#5944]: https://github.com/nymtech/nym/pull/5944
-[#5943]: https://github.com/nymtech/nym/pull/5943
-[#5942]: https://github.com/nymtech/nym/pull/5942
-[#5941]: https://github.com/nymtech/nym/pull/5941
-[#5938]: https://github.com/nymtech/nym/pull/5938
-[#5934]: https://github.com/nymtech/nym/pull/5934
-[#5915]: https://github.com/nymtech/nym/pull/5915
-
 ## [2025.15-gruyere] (2025-08-20)
 
 - Migrate strum to 0.27.2 ([#5960])

Cargo.toml

@@ -31,7 +31,6 @@ members = [
     "common/client-libs/mixnet-client",
     "common/client-libs/validator-client",
     "common/commands",
-    "common/nym-common",
     "common/config",
     "common/cosmwasm-smart-contracts/coconut-dkg",
     "common/cosmwasm-smart-contracts/contracts-common",
@@ -44,7 +43,6 @@ members = [
     "common/cosmwasm-smart-contracts/nym-performance-contract",
     "common/cosmwasm-smart-contracts/nym-pool-contract",
     "common/cosmwasm-smart-contracts/vesting-contract",
-    "common/credential-proxy",
     "common/credential-storage",
     "common/credential-utils",
     "common/credential-verification",
@@ -55,12 +53,12 @@ members = [
     "common/ecash-signer-check",
     "common/ecash-signer-check-types",
     "common/ecash-time",
+    "common/execute",
     "common/exit-policy",
     "common/gateway-requests",
     "common/gateway-stats-storage",
     "common/gateway-storage",
     "common/http-api-client",
-    "common/http-api-client-macro",
     "common/http-api-common",
     "common/inclusion-probability",
     "common/ip-packet-requests",
@@ -69,8 +67,6 @@ members = [
     "common/network-defaults",
     "common/node-tester-utils",
     "common/nonexhaustive-delayqueue",
-    "common/nym-cache",
-    "common/nym-connection-monitor",
     "common/nym-id",
     "common/nym-metrics",
     "common/nym_offline_compact_ecash",
@@ -89,7 +85,6 @@ members = [
     "common/nymsphinx/types",
     "common/nyxd-scraper",
     "common/pemstore",
-    "common/registration",
     "common/serde-helpers",
     "common/service-provider-requests-common",
     "common/socks5-client-core",
@@ -97,13 +92,11 @@ members = [
     "common/socks5/requests",
     "common/statistics",
     "common/store-cipher",
-    "common/task",
-    "common/test-utils",
+    "common/task", "common/test-utils",
     "common/ticketbooks-merkle",
     "common/topology",
     "common/tun",
     "common/types",
-    "common/upgrade-mode-check",
     "common/verloc",
     "common/wasm/client-core",
     "common/wasm/storage",
@@ -133,10 +126,9 @@ members = [
     "nym-node/nym-node-metrics",
     "nym-node/nym-node-requests",
     "nym-outfox",
-    "nym-registration-client",
-    "nym-signers-monitor",
     "nym-statistics-api",
     "nym-validator-rewarder",
+    "nym-wg-gateway-client",
     "nyx-chain-watcher",
     "sdk/ffi/cpp",
     "sdk/ffi/go",
@@ -153,6 +145,7 @@ members = [
     # "tools/internal/sdk-version-bump",
     "tools/internal/ssl-inject",
     "tools/internal/testnet-manager",
+    "tools/internal/testnet-manager",
     "tools/internal/testnet-manager/dkg-bypass-contract",
     "tools/internal/validator-status-check",
     "tools/nym-cli",
@@ -165,7 +158,6 @@ members = [
     "wasm/mix-fetch",
     "wasm/node-tester",
     "wasm/zknym-lib",
-    "nym-gateway-probe"
 ]
 
 default-members = [
@@ -184,16 +176,16 @@ default-members = [
     "tools/nymvisor",
 ]
 
-exclude = ["contracts", "nym-wallet", "cpu-cycles"]
+exclude = ["explorer", "contracts", "nym-wallet", "cpu-cycles"]
 
 [workspace.package]
 authors = ["Nym Technologies SA"]
 repository = "https://github.com/nymtech/nym"
 homepage = "https://nymtech.net"
 documentation = "https://nymtech.net"
-edition = "2024"
+edition = "2021"
 license = "Apache-2.0"
-rust-version = "1.85"
+rust-version = "1.81"
 readme = "README.md"
 
 [workspace.dependencies]
@@ -234,7 +226,7 @@ clap_complete = "4.5"
 clap_complete_fig = "4.5"
 colored = "2.2"
 comfy-table = "7.1.4"
-console = "0.16.0"
+console = "0.15.11"
 console-subscriber = "0.4.1"
 console_error_panic_hook = "0.1"
 const-str = "0.5.6"
@@ -248,7 +240,7 @@ dashmap = "5.5.3"
 # We want https://github.com/DefGuard/wireguard-rs/pull/64 , but there's no crates.io release being pushed out anymore
 defguard_wireguard_rs = { git = "https://github.com/DefGuard/wireguard-rs.git", rev = "v0.4.7" }
 digest = "0.10.7"
-dirs = "6.0"
+dirs = "5.0"
 doc-comment = "0.3"
 dotenvy = "0.15.6"
 dyn-clone = "1.0.19"
@@ -281,13 +273,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"
@@ -300,16 +290,12 @@ 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"
 pin-project = "1.1"
-pnet_packet = "0.35.0"
 pin-project-lite = "0.2.16"
 publicsuffix = "2.3.0"
 proc_pidinfo = "0.1.3"
@@ -335,7 +321,6 @@ 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"
@@ -344,15 +329,14 @@ sqlx = "0.8.6"
 strum = "0.27.2"
 strum_macros = "0.27.2"
 subtle-encoding = "0.5"
-syn = "2"
+syn = "1"
 sysinfo = "0.37.0"
 tap = "1.0.1"
 tar = "0.4.44"
-test-with = { version = "0.15.4", default-features = false }
 tempfile = "3.20"
 thiserror = "2.0"
 time = "0.3.41"
-tokio = "1.47"
+tokio = "1.45"
 tokio-postgres = "0.7"
 tokio-stream = "0.1.17"
 tokio-test = "0.4.4"
@@ -363,14 +347,11 @@ 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"
-tracing-test = "0.2.5"
 ts-rs = "10.1.0"
 tungstenite = { version = "0.20.1", default-features = false }
 uniffi = "0.29.2"
@@ -381,7 +362,6 @@ utoipa-swagger-ui = "8.1"
 utoipauto = "0.2"
 uuid = "*"
 vergen = { version = "=8.3.1", default-features = false }
-vergen-gitcl = { version = "1.0.8", default-features = false }
 walkdir = "2"
 x25519-dalek = "2.0.0"
 zeroize = "1.7.0"
@@ -421,19 +401,18 @@ prost = { version = "0.13", default-features = false }
 # wasm-related dependencies
 gloo-utils = "0.2.0"
 gloo-net = "0.6.0"
-gloo-timers = "0.3.0"
 
 indexed_db_futures = "0.6.4"
 js-sys = "0.3.76"
 serde-wasm-bindgen = "0.6.5"
 tsify = "0.4.5"
-tokio_with_wasm = { version = "0.8.7" }
 wasm-bindgen = "0.2.99"
 wasm-bindgen-futures = "0.4.49"
 wasm-bindgen-test = "0.3.49"
 wasmtimer = "0.4.1"
 web-sys = "0.3.76"
 
+
 # for local development:
 #[patch.crates-io]
 #sphinx-packet = { path = "../sphinx" }

Makefile

@@ -107,16 +107,16 @@ sdk-wasm-build:
 	$(MAKE) -C nym-browser-extension/storage wasm-pack
 	$(MAKE) -C wasm/client
 	$(MAKE) -C wasm/node-tester
-	# $(MAKE) -C wasm/mix-fetch
+	$(MAKE) -C wasm/mix-fetch
 	$(MAKE) -C wasm/zknym-lib
-	# $(MAKE) -C wasm/full-nym-wasm
+	#$(MAKE) -C wasm/full-nym-wasm
 
 # run this from npm/yarn to ensure tools are in the path, e.g. yarn build:sdk from root of repo
 sdk-typescript-build:
 	npx lerna run --scope @nymproject/sdk build --stream
-	# npx lerna run --scope @nymproject/mix-fetch build --stream
-	# npx lerna run --scope @nymproject/node-tester build --stream
-	# yarn --cwd sdk/typescript/codegen/contract-clients build
+	npx lerna run --scope @nymproject/mix-fetch build --stream
+	npx lerna run --scope @nymproject/node-tester build --stream
+	yarn --cwd sdk/typescript/codegen/contract-clients build
 
 # NOTE: These targets are part of the main workspace (but not as wasm32-unknown-unknown)
 WASM_CRATES = extension-storage nym-client-wasm nym-node-tester-wasm zknym-lib
@@ -154,7 +154,6 @@ CONTRACTS_OUT_DIR = contracts/artifacts
 #
 COSMWASM_OPTIMIZER_IMAGE ?= cosmwasm/optimizer:0.17.0
 COSMWASM_OPTIMIZER_PLATFORM ?= linux/amd64
-COSMWASM_CHECK_IMAGE ?= rust:1.88
 
 # Ensure clean build environment and run the optimizer
 optimize-contracts:
@@ -180,13 +179,6 @@ optimize-contracts:
 	# Cleanup temporary artefacts directory
 	@rm -rf artifacts 2>/dev/null || true
 
-# Check artifacts with cosmwasm-check inside the optimizer image
-docker-check-contracts:
-	@docker run --rm --platform $(COSMWASM_OPTIMIZER_PLATFORM) \
-	  -v $(CURDIR):/code --workdir /code \
-	  --entrypoint /bin/sh \
-	  $(COSMWASM_CHECK_IMAGE) -lc 'apt-get update && apt-get install -y --no-install-recommends llvm-dev libclang-dev pkg-config && export PATH="/usr/local/cargo/bin:/usr/local/rustup/bin:$$PATH" && cargo install cosmwasm-check --locked && WASMER_ENGINE=universal WASMER_COMPILER=singlepass cosmwasm-check contracts/artifacts/*.wasm'
-
 wasm-opt-contracts:
 	@for WASM in $(WASM_CONTRACT_DIR)/*.wasm; do \
 	  echo "Running wasm-opt on $$WASM"; \

clients/native/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "nym-client"
-version = "1.1.64"
+version = "1.1.61"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 description = "Implementation of the Nym Client"
 edition = "2021"
-rust-version = "1.85"
+rust-version = "1.70"
 license.workspace = true
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -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",
@@ -70,11 +71,3 @@ nym-client-websocket-requests = { path = "websocket-requests" }
 nym-id = { path = "../../common/nym-id" }
 
 [dev-dependencies]
-
-[features]
-default = []
-otel = [
-    "nym-client-core/otel",
-    "nym-bin-common/otel",
-    "nym-gateway-requests/otel",
-]

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...");
@@ -60,7 +57,6 @@ impl SocketClient {
         let ClientInput {
             connection_command_sender,
             input_sender,
-            ..
         } = client_input;
 
         let ClientOutput {
@@ -81,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> {
@@ -123,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 {
@@ -133,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);
         }
@@ -152,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,16 +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,
-            #[cfg(feature = "otel")]
-            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}");
             }
         }
@@ -223,17 +222,10 @@ impl Handler {
             TransmissionLane::ConnectionId(id)
         });
 
-        let input_msg = InputMessage::new_anonymous(
-            recipient,
-            message,
-            reply_surbs,
-            lane,
-            self.packet_type,
-            #[cfg(feature = "otel")]
-            None,
-        );
+        let input_msg =
+            InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
         if let Err(err) = self.msg_input.send(input_msg).await {
-            if !self.shutdown_token.is_cancelled() {
+            if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send anonymous message to the input buffer: {err}");
             }
         }
@@ -261,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}");
             }
         }
@@ -283,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}");
             }
         }
@@ -402,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() {
@@ -447,6 +436,9 @@ impl Handler {
                         break;
                     }
                 }
+                _ = task_client.recv() => {
+                    log::trace!("Websocket handler: Received shutdown");
+                }
             }
         }
         log::debug!("Websocket handler: Exiting");
@@ -472,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,10 +1,10 @@
 [package]
 name = "nym-socks5-client"
-version = "1.1.64"
+version = "1.1.61"
 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"
-rust-version = "1.85"
+rust-version = "1.70"
 license.workspace = true
 
 [dependencies]
@@ -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",
@@ -53,7 +54,3 @@ nym-validator-client = { path = "../../common/client-libs/validator-client", fea
 [features]
 default = []
 eth = []
-otel = [
-    "nym-socks5-client-core/otel",
-    "nym-bin-common/otel",
-]

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/async-file-watcher/src/lib.rs

@@ -1,8 +1,8 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use futures::StreamExt;
 use futures::channel::mpsc;
+use futures::StreamExt;
 use notify::event::{DataChange, MetadataKind, ModifyKind};
 use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
 use std::collections::HashMap;
@@ -96,10 +96,10 @@ impl AsyncFileWatcher {
         // when testing I was consistently getting two `Modify(Data(Any))` events in quick succession
         // (probably to modify content and metadata).
         // we really only want to propagate one of them
-        if let Some(previous) = self.last_received.get(&event.kind)
-            && now.duration_since(*previous) < self.tick_duration
-        {
-            return false;
+        if let Some(previous) = self.last_received.get(&event.kind) {
+            if now.duration_since(*previous) < self.tick_duration {
+                return false;
+            }
         }
 
         let Some(filters) = &self.filters else {

common/authenticator-requests/Cargo.toml

@@ -13,8 +13,6 @@ base64 = { workspace = true }
 bincode = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
-semver = { workspace = true }
-strum_macros = { workspace = true }
 thiserror = { workspace = true }
 
 nym-credentials-interface = { path = "../credentials-interface" }

common/authenticator-requests/src/client_message.rs

@@ -1,273 +0,0 @@
-// Copyright 2025 Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
-
-use nym_sphinx::addressing::Recipient;
-use nym_wireguard_types::PeerPublicKey;
-
-use crate::{
-    AuthenticatorVersion, Error,
-    latest::registration::IpPair,
-    traits::{FinalMessage, InitMessage, QueryBandwidthMessage, TopUpMessage, Versionable},
-    v2, v3, v4, v5,
-};
-
-// This is very redundant with AuthenticatorRequest and I reckon they could be smooshed.
-// It is a bit out of scope for me at the moment though
-#[derive(Debug)]
-pub enum ClientMessage {
-    Initial(Box<dyn InitMessage + Send + Sync + 'static>),
-    Final(Box<dyn FinalMessage + Send + Sync + 'static>),
-    Query(Box<dyn QueryBandwidthMessage + Send + Sync + 'static>),
-    TopUp(Box<dyn TopUpMessage + Send + Sync + 'static>),
-}
-
-impl ClientMessage {
-    // check if message is wasteful e.g. contains a credential
-    pub fn is_wasteful(&self) -> bool {
-        match self {
-            Self::Final(msg) => msg.credential().is_some(),
-            Self::TopUp(_) => true,
-            Self::Initial(_) | Self::Query(_) => false,
-        }
-    }
-
-    fn version(&self) -> AuthenticatorVersion {
-        match self {
-            ClientMessage::Initial(msg) => msg.version(),
-            ClientMessage::Final(msg) => msg.version(),
-            ClientMessage::Query(msg) => msg.version(),
-            ClientMessage::TopUp(msg) => msg.version(),
-        }
-    }
-
-    pub fn bytes(&self, reply_to: Recipient) -> Result<(Vec<u8>, u64), Error> {
-        match self.version() {
-            AuthenticatorVersion::V1 => Err(Error::UnsupportedVersion),
-            AuthenticatorVersion::V2 => {
-                use v2::{
-                    registration::{ClientMac, FinalMessage, GatewayClient, InitMessage},
-                    request::AuthenticatorRequest,
-                };
-                match self {
-                    ClientMessage::Initial(init_message) => {
-                        let (req, id) = AuthenticatorRequest::new_initial_request(
-                            InitMessage {
-                                pub_key: init_message.pub_key(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Final(final_message) => {
-                        let (req, id) = AuthenticatorRequest::new_final_request(
-                            FinalMessage {
-                                gateway_client: GatewayClient {
-                                    pub_key: final_message.gateway_client_pub_key(),
-                                    private_ip: final_message
-                                        .gateway_client_ipv4()
-                                        .ok_or(Error::UnsupportedMessage)?
-                                        .into(),
-                                    mac: ClientMac::new(final_message.gateway_client_mac()),
-                                },
-                                credential: final_message.credential(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Query(query_message) => {
-                        let (req, id) = AuthenticatorRequest::new_query_request(
-                            query_message.pub_key(),
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    _ => Err(Error::UnsupportedMessage),
-                }
-            }
-            AuthenticatorVersion::V3 => {
-                use v3::{
-                    registration::{ClientMac, FinalMessage, GatewayClient, InitMessage},
-                    request::AuthenticatorRequest,
-                    topup::TopUpMessage,
-                };
-                match self {
-                    ClientMessage::Initial(init_message) => {
-                        let (req, id) = AuthenticatorRequest::new_initial_request(
-                            InitMessage {
-                                pub_key: init_message.pub_key(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Final(final_message) => {
-                        let (req, id) = AuthenticatorRequest::new_final_request(
-                            FinalMessage {
-                                gateway_client: GatewayClient {
-                                    pub_key: final_message.gateway_client_pub_key(),
-                                    private_ip: final_message
-                                        .gateway_client_ipv4()
-                                        .ok_or(Error::UnsupportedMessage)?
-                                        .into(),
-                                    mac: ClientMac::new(final_message.gateway_client_mac()),
-                                },
-                                credential: final_message.credential(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Query(query_message) => {
-                        let (req, id) = AuthenticatorRequest::new_query_request(
-                            query_message.pub_key(),
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::TopUp(top_up_message) => {
-                        let (req, id) = AuthenticatorRequest::new_topup_request(
-                            TopUpMessage {
-                                pub_key: top_up_message.pub_key(),
-                                credential: top_up_message.credential(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                }
-            }
-            AuthenticatorVersion::V4 => {
-                use v4::{
-                    registration::{ClientMac, FinalMessage, GatewayClient, InitMessage},
-                    request::AuthenticatorRequest,
-                    topup::TopUpMessage,
-                };
-                match self {
-                    ClientMessage::Initial(init_message) => {
-                        let (req, id) = AuthenticatorRequest::new_initial_request(
-                            InitMessage {
-                                pub_key: init_message.pub_key(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Final(final_message) => {
-                        let (req, id) = AuthenticatorRequest::new_final_request(
-                            FinalMessage {
-                                gateway_client: GatewayClient {
-                                    pub_key: final_message.gateway_client_pub_key(),
-                                    private_ips: IpPair {
-                                        ipv4: final_message
-                                            .gateway_client_ipv4()
-                                            .ok_or(Error::UnsupportedMessage)?,
-                                        ipv6: final_message
-                                            .gateway_client_ipv6()
-                                            .ok_or(Error::UnsupportedMessage)?,
-                                    }
-                                    .into(),
-                                    mac: ClientMac::new(final_message.gateway_client_mac()),
-                                },
-                                credential: final_message.credential(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Query(query_message) => {
-                        let (req, id) = AuthenticatorRequest::new_query_request(
-                            query_message.pub_key(),
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::TopUp(top_up_message) => {
-                        let (req, id) = AuthenticatorRequest::new_topup_request(
-                            TopUpMessage {
-                                pub_key: top_up_message.pub_key(),
-                                credential: top_up_message.credential(),
-                            },
-                            reply_to,
-                        );
-                        Ok((req.to_bytes()?, id))
-                    }
-                }
-            }
-            AuthenticatorVersion::V5 => {
-                use v5::{
-                    registration::{ClientMac, FinalMessage, GatewayClient, InitMessage},
-                    request::AuthenticatorRequest,
-                    topup::TopUpMessage,
-                };
-                match self {
-                    ClientMessage::Initial(init_message) => {
-                        let (req, id) = AuthenticatorRequest::new_initial_request(InitMessage {
-                            pub_key: init_message.pub_key(),
-                        });
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Final(final_message) => {
-                        let (req, id) = AuthenticatorRequest::new_final_request(FinalMessage {
-                            gateway_client: GatewayClient {
-                                pub_key: final_message.gateway_client_pub_key(),
-                                private_ips: IpPair {
-                                    ipv4: final_message
-                                        .gateway_client_ipv4()
-                                        .ok_or(Error::UnsupportedMessage)?,
-                                    ipv6: final_message
-                                        .gateway_client_ipv6()
-                                        .ok_or(Error::UnsupportedMessage)?,
-                                },
-                                mac: ClientMac::new(final_message.gateway_client_mac()),
-                            },
-                            credential: final_message.credential(),
-                        });
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::Query(query_message) => {
-                        let (req, id) =
-                            AuthenticatorRequest::new_query_request(query_message.pub_key());
-                        Ok((req.to_bytes()?, id))
-                    }
-                    ClientMessage::TopUp(top_up_message) => {
-                        let (req, id) = AuthenticatorRequest::new_topup_request(TopUpMessage {
-                            pub_key: top_up_message.pub_key(),
-                            credential: top_up_message.credential(),
-                        });
-                        Ok((req.to_bytes()?, id))
-                    }
-                }
-            }
-            AuthenticatorVersion::UNKNOWN => Err(Error::UnknownVersion),
-        }
-    }
-
-    pub fn use_surbs(&self) -> bool {
-        use AuthenticatorVersion::*;
-        match self.version() {
-            V1 | V2 | V3 | V4 => false,
-            V5 => true,
-            UNKNOWN => true,
-        }
-    }
-}
-
-// Same comment as above struct
-#[derive(Debug)]
-pub struct QueryMessageImpl {
-    pub pub_key: PeerPublicKey,
-    pub version: AuthenticatorVersion,
-}
-
-impl Versionable for QueryMessageImpl {
-    fn version(&self) -> AuthenticatorVersion {
-        self.version
-    }
-}
-
-impl QueryBandwidthMessage for QueryMessageImpl {
-    fn pub_key(&self) -> PeerPublicKey {
-        self.pub_key
-    }
-}

common/authenticator-requests/src/error.rs

@@ -23,17 +23,6 @@ pub enum Error {
     #[error("conversion: {0}")]
     Conversion(String),
 
-    // TODO add version number for debugging
-    #[error("unknown version number")]
-    UnknownVersion,
-
-    // TODO add version number for debugging
-    #[error("unsupported request version")]
-    UnsupportedVersion,
-
-    #[error("gateway doesn't support this type of message")]
-    UnsupportedMessage,
-
-    #[error(transparent)]
-    Bincode(#[from] bincode::Error),
+    #[error("failed to serialize response packet: {source}")]
+    FailedToSerializeResponsePacket { source: Box<bincode::ErrorKind> },
 }

common/authenticator-requests/src/lib.rs

@@ -1,9 +1,6 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-pub mod client_message;
-pub mod request;
-pub mod response;
 pub mod traits;
 pub mod v1;
 pub mod v2;
@@ -13,13 +10,11 @@ pub mod v5;
 
 mod error;
 mod util;
-mod version;
 
 pub use error::Error;
 pub use v5 as latest;
-pub use version::AuthenticatorVersion;
 
-pub const CURRENT_VERSION: u8 = latest::VERSION;
+pub const CURRENT_VERSION: u8 = 5;
 
 fn make_bincode_serializer() -> impl bincode::Options {
     use bincode::Options;

common/authenticator-requests/src/request.rs

@@ -1,204 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use nym_service_provider_requests_common::{Protocol, ServiceProviderType};
-use nym_sphinx::addressing::Recipient;
-
-use crate::traits::{FinalMessage, InitMessage, QueryBandwidthMessage, TopUpMessage};
-use crate::{v1, v2, v3, v4, v5};
-
-#[derive(Debug)]
-pub enum AuthenticatorRequest {
-    Initial {
-        msg: Box<dyn InitMessage + Send + Sync + 'static>,
-        protocol: Protocol,
-        reply_to: Option<Recipient>,
-        request_id: u64,
-    },
-    Final {
-        msg: Box<dyn FinalMessage + Send + Sync + 'static>,
-        protocol: Protocol,
-        reply_to: Option<Recipient>,
-        request_id: u64,
-    },
-    QueryBandwidth {
-        msg: Box<dyn QueryBandwidthMessage + Send + Sync + 'static>,
-        protocol: Protocol,
-        reply_to: Option<Recipient>,
-        request_id: u64,
-    },
-    TopUpBandwidth {
-        msg: Box<dyn TopUpMessage + Send + Sync + 'static>,
-        protocol: Protocol,
-        reply_to: Option<Recipient>,
-        request_id: u64,
-    },
-}
-
-impl From<v1::request::AuthenticatorRequest> for AuthenticatorRequest {
-    fn from(value: v1::request::AuthenticatorRequest) -> Self {
-        match value.data {
-            v1::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
-                msg: Box::new(init_message),
-                protocol: Protocol {
-                    version: value.version,
-                    service_provider_type: ServiceProviderType::Authenticator,
-                },
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v1::request::AuthenticatorRequestData::Final(gateway_client) => Self::Final {
-                msg: Box::new(gateway_client),
-                protocol: Protocol {
-                    version: value.version,
-                    service_provider_type: ServiceProviderType::Authenticator,
-                },
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v1::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
-                Self::QueryBandwidth {
-                    msg: Box::new(peer_public_key),
-                    protocol: Protocol {
-                        version: value.version,
-                        service_provider_type: ServiceProviderType::Authenticator,
-                    },
-                    reply_to: Some(value.reply_to),
-                    request_id: value.request_id,
-                }
-            }
-        }
-    }
-}
-
-impl From<v2::request::AuthenticatorRequest> for AuthenticatorRequest {
-    fn from(value: v2::request::AuthenticatorRequest) -> Self {
-        match value.data {
-            v2::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
-                msg: Box::new(init_message),
-                protocol: value.protocol,
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v2::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
-                msg: final_message,
-                protocol: value.protocol,
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v2::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
-                Self::QueryBandwidth {
-                    msg: Box::new(peer_public_key),
-                    protocol: value.protocol,
-                    reply_to: Some(value.reply_to),
-                    request_id: value.request_id,
-                }
-            }
-        }
-    }
-}
-
-impl From<v3::request::AuthenticatorRequest> for AuthenticatorRequest {
-    fn from(value: v3::request::AuthenticatorRequest) -> Self {
-        match value.data {
-            v3::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
-                msg: Box::new(init_message),
-                protocol: value.protocol,
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v3::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
-                msg: final_message,
-                protocol: value.protocol,
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v3::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
-                Self::QueryBandwidth {
-                    msg: Box::new(peer_public_key),
-                    protocol: value.protocol,
-                    reply_to: Some(value.reply_to),
-                    request_id: value.request_id,
-                }
-            }
-            v3::request::AuthenticatorRequestData::TopUpBandwidth(top_up_message) => {
-                Self::TopUpBandwidth {
-                    msg: top_up_message,
-                    protocol: value.protocol,
-                    reply_to: Some(value.reply_to),
-                    request_id: value.request_id,
-                }
-            }
-        }
-    }
-}
-
-impl From<v4::request::AuthenticatorRequest> for AuthenticatorRequest {
-    fn from(value: v4::request::AuthenticatorRequest) -> Self {
-        match value.data {
-            v4::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
-                msg: Box::new(init_message),
-                protocol: value.protocol,
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v4::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
-                msg: final_message,
-                protocol: value.protocol,
-                reply_to: Some(value.reply_to),
-                request_id: value.request_id,
-            },
-            v4::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
-                Self::QueryBandwidth {
-                    msg: Box::new(peer_public_key),
-                    protocol: value.protocol,
-                    reply_to: Some(value.reply_to),
-                    request_id: value.request_id,
-                }
-            }
-            v4::request::AuthenticatorRequestData::TopUpBandwidth(top_up_message) => {
-                Self::TopUpBandwidth {
-                    msg: top_up_message,
-                    protocol: value.protocol,
-                    reply_to: Some(value.reply_to),
-                    request_id: value.request_id,
-                }
-            }
-        }
-    }
-}
-
-impl From<v5::request::AuthenticatorRequest> for AuthenticatorRequest {
-    fn from(value: v5::request::AuthenticatorRequest) -> Self {
-        match value.data {
-            v5::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
-                msg: Box::new(init_message),
-                protocol: value.protocol,
-                reply_to: None,
-                request_id: value.request_id,
-            },
-            v5::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
-                msg: final_message,
-                protocol: value.protocol,
-                reply_to: None,
-                request_id: value.request_id,
-            },
-            v5::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
-                Self::QueryBandwidth {
-                    msg: Box::new(peer_public_key),
-                    protocol: value.protocol,
-                    reply_to: None,
-                    request_id: value.request_id,
-                }
-            }
-            v5::request::AuthenticatorRequestData::TopUpBandwidth(top_up_message) => {
-                Self::TopUpBandwidth {
-                    msg: top_up_message,
-                    protocol: value.protocol,
-                    reply_to: None,
-                    request_id: value.request_id,
-                }
-            }
-        }
-    }
-}

common/authenticator-requests/src/response.rs

@@ -1,106 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use crate::traits::{
-    Id, PendingRegistrationResponse, RegisteredResponse, RemainingBandwidthResponse,
-    TopUpBandwidthResponse,
-};
-use crate::{v2, v3, v4, v5};
-
-#[derive(Debug)]
-pub enum AuthenticatorResponse {
-    PendingRegistration(Box<dyn PendingRegistrationResponse + Send + Sync + 'static>),
-    Registered(Box<dyn RegisteredResponse + Send + Sync + 'static>),
-    RemainingBandwidth(Box<dyn RemainingBandwidthResponse + Send + Sync + 'static>),
-    TopUpBandwidth(Box<dyn TopUpBandwidthResponse + Send + Sync + 'static>),
-}
-
-impl Id for AuthenticatorResponse {
-    fn id(&self) -> u64 {
-        match self {
-            AuthenticatorResponse::PendingRegistration(pending_registration_response) => {
-                pending_registration_response.id()
-            }
-            AuthenticatorResponse::Registered(registered_response) => registered_response.id(),
-            AuthenticatorResponse::RemainingBandwidth(remaining_bandwidth_response) => {
-                remaining_bandwidth_response.id()
-            }
-            AuthenticatorResponse::TopUpBandwidth(top_up_bandwidth_response) => {
-                top_up_bandwidth_response.id()
-            }
-        }
-    }
-}
-
-impl From<v2::response::AuthenticatorResponse> for AuthenticatorResponse {
-    fn from(value: v2::response::AuthenticatorResponse) -> Self {
-        match value.data {
-            v2::response::AuthenticatorResponseData::PendingRegistration(
-                pending_registration_response,
-            ) => Self::PendingRegistration(Box::new(pending_registration_response)),
-            v2::response::AuthenticatorResponseData::Registered(registered_response) => {
-                Self::Registered(Box::new(registered_response))
-            }
-            v2::response::AuthenticatorResponseData::RemainingBandwidth(
-                remaining_bandwidth_response,
-            ) => Self::RemainingBandwidth(Box::new(remaining_bandwidth_response)),
-        }
-    }
-}
-
-impl From<v3::response::AuthenticatorResponse> for AuthenticatorResponse {
-    fn from(value: v3::response::AuthenticatorResponse) -> Self {
-        match value.data {
-            v3::response::AuthenticatorResponseData::PendingRegistration(
-                pending_registration_response,
-            ) => Self::PendingRegistration(Box::new(pending_registration_response)),
-            v3::response::AuthenticatorResponseData::Registered(registered_response) => {
-                Self::Registered(Box::new(registered_response))
-            }
-            v3::response::AuthenticatorResponseData::RemainingBandwidth(
-                remaining_bandwidth_response,
-            ) => Self::RemainingBandwidth(Box::new(remaining_bandwidth_response)),
-            v3::response::AuthenticatorResponseData::TopUpBandwidth(top_up_bandwidth_response) => {
-                Self::TopUpBandwidth(Box::new(top_up_bandwidth_response))
-            }
-        }
-    }
-}
-
-impl From<v4::response::AuthenticatorResponse> for AuthenticatorResponse {
-    fn from(value: v4::response::AuthenticatorResponse) -> Self {
-        match value.data {
-            v4::response::AuthenticatorResponseData::PendingRegistration(
-                pending_registration_response,
-            ) => Self::PendingRegistration(Box::new(pending_registration_response)),
-            v4::response::AuthenticatorResponseData::Registered(registered_response) => {
-                Self::Registered(Box::new(registered_response))
-            }
-            v4::response::AuthenticatorResponseData::RemainingBandwidth(
-                remaining_bandwidth_response,
-            ) => Self::RemainingBandwidth(Box::new(remaining_bandwidth_response)),
-            v4::response::AuthenticatorResponseData::TopUpBandwidth(top_up_bandwidth_response) => {
-                Self::TopUpBandwidth(Box::new(top_up_bandwidth_response))
-            }
-        }
-    }
-}
-
-impl From<v5::response::AuthenticatorResponse> for AuthenticatorResponse {
-    fn from(value: v5::response::AuthenticatorResponse) -> Self {
-        match value.data {
-            v5::response::AuthenticatorResponseData::PendingRegistration(
-                pending_registration_response,
-            ) => Self::PendingRegistration(Box::new(pending_registration_response)),
-            v5::response::AuthenticatorResponseData::Registered(registered_response) => {
-                Self::Registered(Box::new(registered_response))
-            }
-            v5::response::AuthenticatorResponseData::RemainingBandwidth(
-                remaining_bandwidth_response,
-            ) => Self::RemainingBandwidth(Box::new(remaining_bandwidth_response)),
-            v5::response::AuthenticatorResponseData::TopUpBandwidth(top_up_bandwidth_response) => {
-                Self::TopUpBandwidth(Box::new(top_up_bandwidth_response))
-            }
-        }
-    }
-}

common/authenticator-requests/src/traits.rs

@@ -1,105 +1,49 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::fmt;
-use std::net::{Ipv4Addr, Ipv6Addr};
-
 use nym_credentials_interface::CredentialSpendingData;
 use nym_crypto::asymmetric::x25519::PrivateKey;
+use nym_service_provider_requests_common::{Protocol, ServiceProviderType};
+use nym_sphinx::addressing::clients::Recipient;
 use nym_wireguard_types::PeerPublicKey;
 
-use crate::latest::registration::IpPair;
-use crate::{AuthenticatorVersion, Error, v1, v2, v3, v4, v5};
+use crate::{
+    v1, v2, v3, v4,
+    v5::{self, registration::IpPair},
+    Error,
+};
 
-pub trait Versionable {
-    fn version(&self) -> AuthenticatorVersion;
+#[derive(Copy, Clone, Debug)]
+pub enum AuthenticatorVersion {
+    V1,
+    V2,
+    V3,
+    V4,
+    V5,
+    UNKNOWN,
 }
 
-impl Versionable for v1::GatewayClient {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V1
+impl From<Protocol> for AuthenticatorVersion {
+    fn from(value: Protocol) -> Self {
+        if value.service_provider_type != ServiceProviderType::Authenticator {
+            AuthenticatorVersion::UNKNOWN
+        } else if value.version == v1::VERSION {
+            AuthenticatorVersion::V1
+        } else if value.version == v2::VERSION {
+            AuthenticatorVersion::V2
+        } else if value.version == v3::VERSION {
+            AuthenticatorVersion::V3
+        } else if value.version == v4::VERSION {
+            AuthenticatorVersion::V4
+        } else if value.version == v5::VERSION {
+            AuthenticatorVersion::V5
+        } else {
+            AuthenticatorVersion::UNKNOWN
+        }
     }
 }
 
-impl Versionable for v1::registration::InitMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V1
-    }
-}
-
-impl Versionable for v2::registration::InitMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V2
-    }
-}
-
-impl Versionable for v3::registration::InitMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V3
-    }
-}
-
-impl Versionable for v4::registration::InitMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V4
-    }
-}
-
-impl Versionable for v5::registration::InitMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V5
-    }
-}
-
-impl Versionable for v2::registration::FinalMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V2
-    }
-}
-
-impl Versionable for v3::registration::FinalMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V3
-    }
-}
-
-impl Versionable for v4::registration::FinalMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V4
-    }
-}
-
-impl Versionable for v5::registration::FinalMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V5
-    }
-}
-
-impl Versionable for PeerPublicKey {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V3
-    }
-}
-
-impl Versionable for v3::topup::TopUpMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V3
-    }
-}
-
-impl Versionable for v4::topup::TopUpMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V4
-    }
-}
-
-impl Versionable for v5::topup::TopUpMessage {
-    fn version(&self) -> AuthenticatorVersion {
-        AuthenticatorVersion::V5
-    }
-}
-
-pub trait InitMessage: Versionable + fmt::Debug {
+pub trait InitMessage {
     fn pub_key(&self) -> PeerPublicKey;
 }
 
@@ -133,18 +77,15 @@ impl InitMessage for v5::registration::InitMessage {
     }
 }
 
-pub trait FinalMessage: Versionable + fmt::Debug {
-    fn gateway_client_pub_key(&self) -> PeerPublicKey;
+pub trait FinalMessage {
+    fn pub_key(&self) -> PeerPublicKey;
     fn verify(&self, private_key: &PrivateKey, nonce: u64) -> Result<(), Error>;
     fn private_ips(&self) -> IpPair;
-    fn gateway_client_ipv4(&self) -> Option<Ipv4Addr>;
-    fn gateway_client_ipv6(&self) -> Option<Ipv6Addr>;
-    fn gateway_client_mac(&self) -> Vec<u8>;
     fn credential(&self) -> Option<CredentialSpendingData>;
 }
 
 impl FinalMessage for v1::GatewayClient {
-    fn gateway_client_pub_key(&self) -> PeerPublicKey {
+    fn pub_key(&self) -> PeerPublicKey {
         self.pub_key
     }
 
@@ -156,28 +97,13 @@ impl FinalMessage for v1::GatewayClient {
         self.private_ip.into()
     }
 
-    fn gateway_client_ipv4(&self) -> Option<Ipv4Addr> {
-        match self.private_ip {
-            std::net::IpAddr::V4(ipv4_addr) => Some(ipv4_addr),
-            std::net::IpAddr::V6(_) => None,
-        }
-    }
-
-    fn gateway_client_ipv6(&self) -> Option<Ipv6Addr> {
-        None
-    }
-
-    fn gateway_client_mac(&self) -> Vec<u8> {
-        self.mac.to_vec()
-    }
-
     fn credential(&self) -> Option<CredentialSpendingData> {
         None
     }
 }
 
 impl FinalMessage for v2::registration::FinalMessage {
-    fn gateway_client_pub_key(&self) -> PeerPublicKey {
+    fn pub_key(&self) -> PeerPublicKey {
         self.gateway_client.pub_key
     }
 
@@ -189,28 +115,13 @@ impl FinalMessage for v2::registration::FinalMessage {
         self.gateway_client.private_ip.into()
     }
 
-    fn gateway_client_ipv4(&self) -> Option<Ipv4Addr> {
-        match self.gateway_client.private_ip {
-            std::net::IpAddr::V4(ipv4_addr) => Some(ipv4_addr),
-            std::net::IpAddr::V6(_) => None,
-        }
-    }
-
-    fn gateway_client_ipv6(&self) -> Option<Ipv6Addr> {
-        None
-    }
-
-    fn gateway_client_mac(&self) -> Vec<u8> {
-        self.gateway_client.mac.to_vec()
-    }
-
     fn credential(&self) -> Option<CredentialSpendingData> {
         self.credential.clone()
     }
 }
 
 impl FinalMessage for v3::registration::FinalMessage {
-    fn gateway_client_pub_key(&self) -> PeerPublicKey {
+    fn pub_key(&self) -> PeerPublicKey {
         self.gateway_client.pub_key
     }
 
@@ -222,28 +133,13 @@ impl FinalMessage for v3::registration::FinalMessage {
         self.gateway_client.private_ip.into()
     }
 
-    fn gateway_client_ipv4(&self) -> Option<Ipv4Addr> {
-        match self.gateway_client.private_ip {
-            std::net::IpAddr::V4(ipv4_addr) => Some(ipv4_addr),
-            std::net::IpAddr::V6(_) => None,
-        }
-    }
-
-    fn gateway_client_ipv6(&self) -> Option<Ipv6Addr> {
-        None
-    }
-
-    fn gateway_client_mac(&self) -> Vec<u8> {
-        self.gateway_client.mac.to_vec()
-    }
-
     fn credential(&self) -> Option<CredentialSpendingData> {
         self.credential.clone()
     }
 }
 
 impl FinalMessage for v4::registration::FinalMessage {
-    fn gateway_client_pub_key(&self) -> PeerPublicKey {
+    fn pub_key(&self) -> PeerPublicKey {
         self.gateway_client.pub_key
     }
 
@@ -255,25 +151,13 @@ impl FinalMessage for v4::registration::FinalMessage {
         self.gateway_client.private_ips.into()
     }
 
-    fn gateway_client_ipv4(&self) -> Option<Ipv4Addr> {
-        Some(self.gateway_client.private_ips.ipv4)
-    }
-
-    fn gateway_client_ipv6(&self) -> Option<Ipv6Addr> {
-        Some(self.gateway_client.private_ips.ipv6)
-    }
-
-    fn gateway_client_mac(&self) -> Vec<u8> {
-        self.gateway_client.mac.to_vec()
-    }
-
     fn credential(&self) -> Option<CredentialSpendingData> {
         self.credential.clone()
     }
 }
 
 impl FinalMessage for v5::registration::FinalMessage {
-    fn gateway_client_pub_key(&self) -> PeerPublicKey {
+    fn pub_key(&self) -> PeerPublicKey {
         self.gateway_client.pub_key
     }
 
@@ -285,24 +169,12 @@ impl FinalMessage for v5::registration::FinalMessage {
         self.gateway_client.private_ips
     }
 
-    fn gateway_client_ipv4(&self) -> Option<Ipv4Addr> {
-        Some(self.gateway_client.private_ips.ipv4)
-    }
-
-    fn gateway_client_ipv6(&self) -> Option<Ipv6Addr> {
-        Some(self.gateway_client.private_ips.ipv6)
-    }
-
-    fn gateway_client_mac(&self) -> Vec<u8> {
-        self.gateway_client.mac.to_vec()
-    }
-
     fn credential(&self) -> Option<CredentialSpendingData> {
         self.credential.clone()
     }
 }
 
-pub trait QueryBandwidthMessage: Versionable + fmt::Debug {
+pub trait QueryBandwidthMessage {
     fn pub_key(&self) -> PeerPublicKey;
 }
 
@@ -312,7 +184,7 @@ impl QueryBandwidthMessage for PeerPublicKey {
     }
 }
 
-pub trait TopUpMessage: Versionable + fmt::Debug {
+pub trait TopUpMessage {
     fn pub_key(&self) -> PeerPublicKey;
     fn credential(&self) -> CredentialSpendingData;
 }
@@ -347,286 +219,197 @@ impl TopUpMessage for v5::topup::TopUpMessage {
     }
 }
 
-pub trait Id {
-    fn id(&self) -> u64;
+pub enum AuthenticatorRequest {
+    Initial {
+        msg: Box<dyn InitMessage + Send + Sync + 'static>,
+        protocol: Protocol,
+        reply_to: Option<Recipient>,
+        request_id: u64,
+    },
+    Final {
+        msg: Box<dyn FinalMessage + Send + Sync + 'static>,
+        protocol: Protocol,
+        reply_to: Option<Recipient>,
+        request_id: u64,
+    },
+    QueryBandwidth {
+        msg: Box<dyn QueryBandwidthMessage + Send + Sync + 'static>,
+        protocol: Protocol,
+        reply_to: Option<Recipient>,
+        request_id: u64,
+    },
+    TopUpBandwidth {
+        msg: Box<dyn TopUpMessage + Send + Sync + 'static>,
+        protocol: Protocol,
+        reply_to: Option<Recipient>,
+        request_id: u64,
+    },
 }
 
-impl Id for v2::response::PendingRegistrationResponse {
-    fn id(&self) -> u64 {
-        self.request_id
+impl From<v1::request::AuthenticatorRequest> for AuthenticatorRequest {
+    fn from(value: v1::request::AuthenticatorRequest) -> Self {
+        match value.data {
+            v1::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
+                msg: Box::new(init_message),
+                protocol: Protocol {
+                    version: value.version,
+                    service_provider_type: ServiceProviderType::Authenticator,
+                },
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v1::request::AuthenticatorRequestData::Final(gateway_client) => Self::Final {
+                msg: Box::new(gateway_client),
+                protocol: Protocol {
+                    version: value.version,
+                    service_provider_type: ServiceProviderType::Authenticator,
+                },
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v1::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
+                Self::QueryBandwidth {
+                    msg: Box::new(peer_public_key),
+                    protocol: Protocol {
+                        version: value.version,
+                        service_provider_type: ServiceProviderType::Authenticator,
+                    },
+                    reply_to: Some(value.reply_to),
+                    request_id: value.request_id,
+                }
+            }
+        }
     }
 }
 
-impl Id for v3::response::PendingRegistrationResponse {
-    fn id(&self) -> u64 {
-        self.request_id
+impl From<v2::request::AuthenticatorRequest> for AuthenticatorRequest {
+    fn from(value: v2::request::AuthenticatorRequest) -> Self {
+        match value.data {
+            v2::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
+                msg: Box::new(init_message),
+                protocol: value.protocol,
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v2::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
+                msg: final_message,
+                protocol: value.protocol,
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v2::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
+                Self::QueryBandwidth {
+                    msg: Box::new(peer_public_key),
+                    protocol: value.protocol,
+                    reply_to: Some(value.reply_to),
+                    request_id: value.request_id,
+                }
+            }
+        }
     }
 }
 
-impl Id for v4::response::PendingRegistrationResponse {
-    fn id(&self) -> u64 {
-        self.request_id
+impl From<v3::request::AuthenticatorRequest> for AuthenticatorRequest {
+    fn from(value: v3::request::AuthenticatorRequest) -> Self {
+        match value.data {
+            v3::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
+                msg: Box::new(init_message),
+                protocol: value.protocol,
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v3::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
+                msg: final_message,
+                protocol: value.protocol,
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v3::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
+                Self::QueryBandwidth {
+                    msg: Box::new(peer_public_key),
+                    protocol: value.protocol,
+                    reply_to: Some(value.reply_to),
+                    request_id: value.request_id,
+                }
+            }
+            v3::request::AuthenticatorRequestData::TopUpBandwidth(top_up_message) => {
+                Self::TopUpBandwidth {
+                    msg: top_up_message,
+                    protocol: value.protocol,
+                    reply_to: Some(value.reply_to),
+                    request_id: value.request_id,
+                }
+            }
+        }
     }
 }
 
-impl Id for v5::response::PendingRegistrationResponse {
-    fn id(&self) -> u64 {
-        self.request_id
+impl From<v4::request::AuthenticatorRequest> for AuthenticatorRequest {
+    fn from(value: v4::request::AuthenticatorRequest) -> Self {
+        match value.data {
+            v4::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
+                msg: Box::new(init_message),
+                protocol: value.protocol,
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v4::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
+                msg: final_message,
+                protocol: value.protocol,
+                reply_to: Some(value.reply_to),
+                request_id: value.request_id,
+            },
+            v4::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
+                Self::QueryBandwidth {
+                    msg: Box::new(peer_public_key),
+                    protocol: value.protocol,
+                    reply_to: Some(value.reply_to),
+                    request_id: value.request_id,
+                }
+            }
+            v4::request::AuthenticatorRequestData::TopUpBandwidth(top_up_message) => {
+                Self::TopUpBandwidth {
+                    msg: top_up_message,
+                    protocol: value.protocol,
+                    reply_to: Some(value.reply_to),
+                    request_id: value.request_id,
+                }
+            }
+        }
     }
 }
 
-impl Id for v2::response::RegisteredResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v3::response::RegisteredResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v4::response::RegisteredResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v5::response::RegisteredResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v2::response::RemainingBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v3::response::RemainingBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v4::response::RemainingBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v5::response::RemainingBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v3::response::TopUpBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v4::response::TopUpBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-impl Id for v5::response::TopUpBandwidthResponse {
-    fn id(&self) -> u64 {
-        self.request_id
-    }
-}
-
-pub trait PendingRegistrationResponse: Id + fmt::Debug {
-    fn nonce(&self) -> u64;
-    fn verify(&self, gateway_key: &PrivateKey) -> std::result::Result<(), Error>;
-    fn pub_key(&self) -> PeerPublicKey;
-    fn private_ips(&self) -> IpPair;
-}
-
-impl PendingRegistrationResponse for v2::response::PendingRegistrationResponse {
-    fn nonce(&self) -> u64 {
-        self.reply.nonce
-    }
-
-    fn verify(&self, gateway_key: &PrivateKey) -> std::result::Result<(), Error> {
-        self.reply.gateway_data.verify(gateway_key, self.nonce())
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.gateway_data.pub_key
-    }
-
-    fn private_ips(&self) -> IpPair {
-        self.reply.gateway_data.private_ip.into()
-    }
-}
-
-impl PendingRegistrationResponse for v3::response::PendingRegistrationResponse {
-    fn nonce(&self) -> u64 {
-        self.reply.nonce
-    }
-
-    fn verify(&self, gateway_key: &PrivateKey) -> std::result::Result<(), Error> {
-        self.reply.gateway_data.verify(gateway_key, self.nonce())
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.gateway_data.pub_key
-    }
-
-    fn private_ips(&self) -> IpPair {
-        self.reply.gateway_data.private_ip.into()
-    }
-}
-
-impl PendingRegistrationResponse for v4::response::PendingRegistrationResponse {
-    fn nonce(&self) -> u64 {
-        self.reply.nonce
-    }
-
-    fn verify(&self, gateway_key: &PrivateKey) -> std::result::Result<(), Error> {
-        self.reply.gateway_data.verify(gateway_key, self.nonce())
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.gateway_data.pub_key
-    }
-
-    fn private_ips(&self) -> IpPair {
-        self.reply.gateway_data.private_ips.into()
-    }
-}
-
-impl PendingRegistrationResponse for v5::response::PendingRegistrationResponse {
-    fn nonce(&self) -> u64 {
-        self.reply.nonce
-    }
-
-    fn verify(&self, gateway_key: &PrivateKey) -> std::result::Result<(), Error> {
-        self.reply.gateway_data.verify(gateway_key, self.nonce())
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.gateway_data.pub_key
-    }
-
-    fn private_ips(&self) -> IpPair {
-        self.reply.gateway_data.private_ips
-    }
-}
-
-pub trait RegisteredResponse: Id + fmt::Debug {
-    fn private_ips(&self) -> IpPair;
-    fn pub_key(&self) -> PeerPublicKey;
-    fn wg_port(&self) -> u16;
-}
-
-impl RegisteredResponse for v2::response::RegisteredResponse {
-    fn private_ips(&self) -> IpPair {
-        self.reply.private_ip.into()
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.pub_key
-    }
-
-    fn wg_port(&self) -> u16 {
-        self.reply.wg_port
-    }
-}
-
-impl RegisteredResponse for v3::response::RegisteredResponse {
-    fn private_ips(&self) -> IpPair {
-        self.reply.private_ip.into()
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.pub_key
-    }
-
-    fn wg_port(&self) -> u16 {
-        self.reply.wg_port
-    }
-}
-impl RegisteredResponse for v4::response::RegisteredResponse {
-    fn private_ips(&self) -> IpPair {
-        self.reply.private_ips.into()
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.pub_key
-    }
-
-    fn wg_port(&self) -> u16 {
-        self.reply.wg_port
-    }
-}
-
-impl RegisteredResponse for v5::response::RegisteredResponse {
-    fn private_ips(&self) -> IpPair {
-        self.reply.private_ips
-    }
-
-    fn pub_key(&self) -> PeerPublicKey {
-        self.reply.pub_key
-    }
-
-    fn wg_port(&self) -> u16 {
-        self.reply.wg_port
-    }
-}
-
-pub trait RemainingBandwidthResponse: Id + fmt::Debug {
-    fn available_bandwidth(&self) -> Option<i64>;
-}
-
-impl RemainingBandwidthResponse for v2::response::RemainingBandwidthResponse {
-    fn available_bandwidth(&self) -> Option<i64> {
-        self.reply.as_ref().map(|r| r.available_bandwidth)
-    }
-}
-
-impl RemainingBandwidthResponse for v3::response::RemainingBandwidthResponse {
-    fn available_bandwidth(&self) -> Option<i64> {
-        self.reply.as_ref().map(|r| r.available_bandwidth)
-    }
-}
-
-impl RemainingBandwidthResponse for v4::response::RemainingBandwidthResponse {
-    fn available_bandwidth(&self) -> Option<i64> {
-        self.reply.as_ref().map(|r| r.available_bandwidth)
-    }
-}
-
-impl RemainingBandwidthResponse for v5::response::RemainingBandwidthResponse {
-    fn available_bandwidth(&self) -> Option<i64> {
-        self.reply.as_ref().map(|r| r.available_bandwidth)
-    }
-}
-
-pub trait TopUpBandwidthResponse: Id + fmt::Debug {
-    fn available_bandwidth(&self) -> i64;
-}
-
-impl TopUpBandwidthResponse for v3::response::TopUpBandwidthResponse {
-    fn available_bandwidth(&self) -> i64 {
-        self.reply.available_bandwidth
-    }
-}
-
-impl TopUpBandwidthResponse for v4::response::TopUpBandwidthResponse {
-    fn available_bandwidth(&self) -> i64 {
-        self.reply.available_bandwidth
-    }
-}
-
-impl TopUpBandwidthResponse for v5::response::TopUpBandwidthResponse {
-    fn available_bandwidth(&self) -> i64 {
-        self.reply.available_bandwidth
+impl From<v5::request::AuthenticatorRequest> for AuthenticatorRequest {
+    fn from(value: v5::request::AuthenticatorRequest) -> Self {
+        match value.data {
+            v5::request::AuthenticatorRequestData::Initial(init_message) => Self::Initial {
+                msg: Box::new(init_message),
+                protocol: value.protocol,
+                reply_to: None,
+                request_id: value.request_id,
+            },
+            v5::request::AuthenticatorRequestData::Final(final_message) => Self::Final {
+                msg: final_message,
+                protocol: value.protocol,
+                reply_to: None,
+                request_id: value.request_id,
+            },
+            v5::request::AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
+                Self::QueryBandwidth {
+                    msg: Box::new(peer_public_key),
+                    protocol: value.protocol,
+                    reply_to: None,
+                    request_id: value.request_id,
+                }
+            }
+            v5::request::AuthenticatorRequestData::TopUpBandwidth(top_up_message) => {
+                Self::TopUpBandwidth {
+                    msg: top_up_message,
+                    protocol: value.protocol,
+                    reply_to: None,
+                    request_id: value.request_id,
+                }
+            }
+        }
     }
 }

common/authenticator-requests/src/v1/registration.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::Error;
-use base64::{Engine, engine::general_purpose};
+use base64::{engine::general_purpose, Engine};
 use nym_wireguard_types::PeerPublicKey;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;

common/authenticator-requests/src/v2/registration.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::Error;
-use base64::{Engine, engine::general_purpose};
+use base64::{engine::general_purpose, Engine};
 use nym_credentials_interface::CredentialSpendingData;
 use nym_wireguard_types::PeerPublicKey;
 use serde::{Deserialize, Serialize};

common/authenticator-requests/src/v3/registration.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::Error;
-use base64::{Engine, engine::general_purpose};
+use base64::{engine::general_purpose, Engine};
 use nym_credentials_interface::CredentialSpendingData;
 use nym_wireguard_types::PeerPublicKey;
 use serde::{Deserialize, Serialize};

common/authenticator-requests/src/v4/registration.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::Error;
-use base64::{Engine, engine::general_purpose};
+use base64::{engine::general_purpose, Engine};
 use nym_credentials_interface::CredentialSpendingData;
 use nym_network_defaults::constants::{WG_TUN_DEVICE_IP_ADDRESS_V4, WG_TUN_DEVICE_IP_ADDRESS_V6};
 use nym_wireguard_types::PeerPublicKey;

common/authenticator-requests/src/v5/registration.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::Error;
-use base64::{Engine, engine::general_purpose};
+use base64::{engine::general_purpose, Engine};
 use nym_credentials_interface::CredentialSpendingData;
 use nym_network_defaults::constants::{WG_TUN_DEVICE_IP_ADDRESS_V4, WG_TUN_DEVICE_IP_ADDRESS_V6};
 use nym_wireguard_types::PeerPublicKey;

common/authenticator-requests/src/version.rs

@@ -1,195 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use super::{v1, v2, v3, v4, v5};
-use nym_service_provider_requests_common::{Protocol, ServiceProviderType};
-
-#[derive(Copy, Clone, Debug, PartialEq, strum_macros::Display)]
-#[strum(serialize_all = "snake_case")]
-pub enum AuthenticatorVersion {
-    /// introduced in wispa release (1.1.5)
-    V1,
-
-    /// introduced in aero release (1.1.9)
-    V2,
-
-    /// introduced in magura release (1.1.10)
-    V3,
-
-    /// introduced in crunch release (1.2.0)
-    V4,
-
-    /// introduced in dorina-patched release (1.6.1)
-    V5,
-
-    UNKNOWN,
-}
-
-impl AuthenticatorVersion {
-    pub const LATEST: Self = Self::V5;
-
-    pub const fn release_version(&self) -> semver::Version {
-        match self {
-            AuthenticatorVersion::V1 => semver::Version::new(1, 1, 5),
-            AuthenticatorVersion::V2 => semver::Version::new(1, 1, 9),
-            AuthenticatorVersion::V3 => semver::Version::new(1, 1, 10),
-            AuthenticatorVersion::V4 => semver::Version::new(1, 2, 0),
-            AuthenticatorVersion::V5 => semver::Version::new(1, 6, 1),
-            AuthenticatorVersion::UNKNOWN => semver::Version::new(0, 0, 0),
-        }
-    }
-}
-
-impl From<Protocol> for AuthenticatorVersion {
-    fn from(value: Protocol) -> Self {
-        if value.service_provider_type != ServiceProviderType::Authenticator {
-            AuthenticatorVersion::UNKNOWN
-        } else if value.version == v1::VERSION {
-            AuthenticatorVersion::V1
-        } else if value.version == v2::VERSION {
-            AuthenticatorVersion::V2
-        } else if value.version == v3::VERSION {
-            AuthenticatorVersion::V3
-        } else if value.version == v4::VERSION {
-            AuthenticatorVersion::V4
-        } else if value.version == v5::VERSION {
-            AuthenticatorVersion::V5
-        } else {
-            AuthenticatorVersion::UNKNOWN
-        }
-    }
-}
-
-impl From<u8> for AuthenticatorVersion {
-    fn from(value: u8) -> Self {
-        if value == v1::VERSION {
-            AuthenticatorVersion::V1
-        } else if value == v2::VERSION {
-            AuthenticatorVersion::V2
-        } else if value == v3::VERSION {
-            AuthenticatorVersion::V3
-        } else if value == v4::VERSION {
-            AuthenticatorVersion::V4
-        } else if value == v5::VERSION {
-            AuthenticatorVersion::V5
-        } else {
-            AuthenticatorVersion::UNKNOWN
-        }
-    }
-}
-
-impl From<&str> for AuthenticatorVersion {
-    fn from(value: &str) -> Self {
-        let Ok(semver) = semver::Version::parse(value) else {
-            return Self::UNKNOWN;
-        };
-
-        semver.into()
-    }
-}
-
-impl From<Option<&String>> for AuthenticatorVersion {
-    fn from(value: Option<&String>) -> Self {
-        match value {
-            None => Self::UNKNOWN,
-            Some(value) => value.as_str().into(),
-        }
-    }
-}
-
-impl From<String> for AuthenticatorVersion {
-    fn from(value: String) -> Self {
-        Self::from(value.as_str())
-    }
-}
-
-impl From<Option<String>> for AuthenticatorVersion {
-    fn from(value: Option<String>) -> Self {
-        value.as_ref().into()
-    }
-}
-
-impl From<semver::Version> for AuthenticatorVersion {
-    fn from(semver: semver::Version) -> Self {
-        if semver < AuthenticatorVersion::V1.release_version() {
-            return Self::UNKNOWN;
-        }
-        if semver < AuthenticatorVersion::V2.release_version() {
-            return Self::V1;
-        }
-        if semver < AuthenticatorVersion::V3.release_version() {
-            return Self::V2;
-        }
-        if semver < AuthenticatorVersion::V4.release_version() {
-            return Self::V3;
-        }
-        if semver < AuthenticatorVersion::V5.release_version() {
-            return Self::V4;
-        }
-        // if provided version is higher (or equal) to release version of V5,
-        // we return the latest (i.e. v5)
-
-        debug_assert_eq!(
-            Self::V5,
-            Self::LATEST,
-            "a new AuthenticatorVersion variant has been introduced without adjusting the `From<semver::Version>` trait"
-        );
-        Self::LATEST
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::super::latest;
-    use super::*;
-
-    #[test]
-    fn strum_display() {
-        // sanity check on formatting and casing
-        assert_eq!("v1", AuthenticatorVersion::V1.to_string());
-        assert_eq!("v2", AuthenticatorVersion::V2.to_string());
-        assert_eq!("unknown", AuthenticatorVersion::UNKNOWN.to_string());
-    }
-
-    #[test]
-    fn u8_conversion() {
-        assert_eq!(AuthenticatorVersion::V1, AuthenticatorVersion::from(1u8));
-        assert_eq!(AuthenticatorVersion::V2, AuthenticatorVersion::from(2u8));
-
-        assert_eq!(
-            AuthenticatorVersion::UNKNOWN,
-            AuthenticatorVersion::from(latest::VERSION + 1)
-        );
-        assert_eq!(
-            AuthenticatorVersion::UNKNOWN,
-            AuthenticatorVersion::from(0u8)
-        );
-        assert_eq!(
-            AuthenticatorVersion::UNKNOWN,
-            AuthenticatorVersion::from(255u8)
-        );
-    }
-
-    #[test]
-    fn semver_checks() {
-        assert_eq!(AuthenticatorVersion::UNKNOWN, "1.1.4".into());
-        assert_eq!(AuthenticatorVersion::UNKNOWN, "0.1.0".into());
-        assert_eq!(AuthenticatorVersion::UNKNOWN, "1.0.4".into());
-        assert_eq!(AuthenticatorVersion::V1, "1.1.5".into());
-        assert_eq!(AuthenticatorVersion::V1, "1.1.6".into());
-        assert_eq!(AuthenticatorVersion::V1, "1.1.8".into());
-        assert_eq!(AuthenticatorVersion::V2, "1.1.9".into());
-        assert_eq!(AuthenticatorVersion::V3, "1.1.10".into());
-        assert_eq!(AuthenticatorVersion::V3, "1.1.11".into());
-        assert_eq!(AuthenticatorVersion::V3, "1.1.60".into());
-        assert_eq!(AuthenticatorVersion::V4, "1.2.0".into());
-        assert_eq!(AuthenticatorVersion::V4, "1.2.1".into());
-        assert_eq!(AuthenticatorVersion::V4, "1.5.1".into());
-        assert_eq!(AuthenticatorVersion::V4, "1.6.0".into());
-        assert_eq!(AuthenticatorVersion::V5, "1.6.1".into());
-        assert_eq!(AuthenticatorVersion::V5, "1.6.11".into());
-        assert_eq!(AuthenticatorVersion::V5, "1.7.0".into());
-        assert_eq!(AuthenticatorVersion::V5, "1.16.11".into());
-        assert_eq!(AuthenticatorVersion::V5, "1.17.0".into());
-    }
-}

common/bandwidth-controller/Cargo.toml

@@ -7,7 +7,6 @@ license.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-trait = { workspace =  true }
 bip39 = { workspace = true }
 log = { workspace = true }
 rand = { workspace = true }

common/bandwidth-controller/src/lib.rs

@@ -23,12 +23,10 @@ use nym_validator_client::nym_api::EpochId;
 use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
 
 pub use event::BandwidthStatusMessage;
-pub use traits::{BandwidthTicketProvider, DEFAULT_TICKETS_TO_SPEND};
 
 pub mod acquire;
 pub mod error;
 mod event;
-mod traits;
 mod utils;
 
 #[derive(Debug)]

common/bandwidth-controller/src/traits.rs

@@ -1,42 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use async_trait::async_trait;
-use nym_credential_storage::storage::Storage;
-use nym_credentials_interface::TicketType;
-use nym_crypto::asymmetric::ed25519;
-use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
-
-use crate::{error::BandwidthControllerError, BandwidthController, PreparedCredential};
-
-pub const DEFAULT_TICKETS_TO_SPEND: u32 = 1;
-
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-pub trait BandwidthTicketProvider: Send + Sync {
-    async fn get_ecash_ticket(
-        &self,
-        ticket_type: TicketType,
-        gateway_id: ed25519::PublicKey,
-        tickets_to_spend: u32,
-    ) -> Result<PreparedCredential, BandwidthControllerError>;
-}
-
-#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
-#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-impl<C, St> BandwidthTicketProvider for BandwidthController<C, St>
-where
-    C: DkgQueryClient + Sync + Send,
-    St: nym_credential_storage::storage::Storage,
-    <St as Storage>::StorageError: Send + Sync + 'static,
-{
-    async fn get_ecash_ticket(
-        &self,
-        ticket_type: TicketType,
-        gateway_id: ed25519::PublicKey,
-        tickets_to_spend: u32,
-    ) -> Result<PreparedCredential, BandwidthControllerError> {
-        self.prepare_ecash_ticket(ticket_type, gateway_id.to_bytes(), tickets_to_spend)
-            .await
-    }
-}

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/completions/mod.rs

@@ -1,8 +1,8 @@
-use clap::Args;
 use clap::builder::Command;
 use clap::clap_derive::ValueEnum;
-use clap_complete::Shell as ClapShell;
+use clap::Args;
 use clap_complete::generator::generate;
+use clap_complete::Shell as ClapShell;
 use std::io;
 
 pub fn fig_generate(command: &mut Command, name: &str) {

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::trace::{SpanId, TraceId};
-use opentelemetry_sdk::trace::IdGenerator;
-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,155 +0,0 @@
-use opentelemetry::propagation::{Extractor, Injector, TextMapPropagator};
-use opentelemetry::trace::{SpanContext, TraceContextExt, TraceId};
-use opentelemetry::{Context, TraceFlags};
-use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::IdGenerator};
-use std::collections::HashMap;
-use std::fmt::Display;
-use tracing::instrument;
-use tracing_opentelemetry::OpenTelemetrySpanExt;
-
-/// 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,186 +0,0 @@
-pub mod compact_id_generator;
-pub mod context;
-mod trace_id_format;
-
-use tracing::{Level, info};
-use tracing_subscriber::filter::Directive;
-use tracing_subscriber::fmt;
-use tracing_subscriber::layer::SubscriberExt;
-use tracing_subscriber::util::SubscriberInitExt;
-
-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::{KeyValue, global};
-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::{Resource, trace::Sampler};
-use opentelemetry_semantic_conventions::SCHEMA_URL;
-use opentelemetry_semantic_conventions::resource::{DEPLOYMENT_ENVIRONMENT_NAME, SERVICE_VERSION};
-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)
-}

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::AsSerde;
-use tracing_serde::fields::AsMap;
-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

@@ -3,7 +3,7 @@ name = "nym-client-core"
 version = "1.1.15"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 edition = "2021"
-rust-version = "1.85"
+rust-version = "1.76"
 license.workspace = true
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -69,6 +69,7 @@ workspace = true
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.hyper-util]
 workspace = true
 features = ["tokio"]
+###
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-stream]
 workspace = true
@@ -102,7 +103,7 @@ workspace = true
 features = ["tokio"]
 
 [target."cfg(target_arch = \"wasm32\")".dependencies.gloo-timers]
-workspace = true
+version = "0.3.0"
 features = ["futures"]
 
 [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-utils]
@@ -113,10 +114,6 @@ features = ["websocket"]
 workspace = true
 features = ["wasm-bindgen"]
 
-[target."cfg(target_arch = \"wasm32\")".dependencies.tokio_with_wasm]
-workspace = true
-features = ["full"]
-
 [dev-dependencies]
 tempfile = { workspace = true }
 
@@ -126,7 +123,6 @@ 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 = []
 

common/client-core/config-types/src/lib.rs

@@ -707,10 +707,13 @@ pub struct DebugConfig {
 
     /// Defines all configuration options related to reply SURBs.
     pub reply_surbs: ReplySurbs,
+
     /// Defines all configuration options related to stats reporting.
     pub stats_reporting: StatsReporting,
+
     /// Defines all configuration options related to the forget me flag.
     pub forget_me: ForgetMe,
+
     /// Defines all configuration options related to the remember me flag.
     pub remember_me: RememberMe,
 }

common/client-core/config-types/src/old/v6.rs

@@ -543,8 +543,10 @@ pub struct DebugConfigV6 {
 
     /// Defines all configuration options related to reply SURBs.
     pub reply_surbs: ReplySurbsV6,
+
     /// Defines all configuration options related to stats reporting.
     pub stats_reporting: StatsReportingV6,
+
     /// Defines all configuration options related to the forget me flag.
     pub forget_me: ForgetMeV6,
 

common/client-core/gateways-storage/src/backend/fs_backend/manager.rs

@@ -31,6 +31,7 @@ impl StorageManager {
                 }
             })?;
         }
+
         let opts = sqlx::sqlite::SqliteConnectOptions::new()
             .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
             .synchronous(SqliteSynchronous::Normal)

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

@@ -114,12 +114,13 @@ where
         })?;
         hardcoded_topology.entry_capable_nodes().cloned().collect()
     } else {
+        let mut rng = rand::thread_rng();
         crate::init::helpers::gateways_for_init(
+            &mut rng,
             &core.client.nym_api_urls,
             user_agent,
             core.debug.topology.minimum_gateway_performance,
             core.debug.topology.ignore_ingress_epoch_role,
-            None,
         )
         .await?
     };

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

@@ -173,12 +173,13 @@ where
         })?;
         hardcoded_topology.entry_capable_nodes().cloned().collect()
     } else {
+        let mut rng = rand::thread_rng();
         crate::init::helpers::gateways_for_init(
+            &mut rng,
             &core.client.nym_api_urls,
             user_agent,
             core.debug.topology.minimum_gateway_performance,
             core.debug.topology.ignore_ingress_epoch_role,
-            None,
         )
         .await?
     };

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

@@ -7,12 +7,11 @@ use super::statistics_control::StatisticsControl;
 use crate::client::base_client::storage::helpers::store_client_keys;
 use crate::client::base_client::storage::MixnetClientStorage;
 use crate::client::cover_traffic_stream::LoopCoverTrafficStream;
-use crate::client::event_control::EventControl;
 use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender};
 use crate::client::key_manager::persistence::KeyStore;
 use crate::client::key_manager::ClientKeys;
 use crate::client::mix_traffic::transceiver::{GatewayReceiver, GatewayTransceiver, RemoteGateway};
-use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController, MixTrafficEvent};
+use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
 use crate::client::real_messages_control;
 use crate::client::real_messages_control::RealMessagesController;
 use crate::client::received_buffer::{
@@ -28,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};
@@ -49,11 +48,12 @@ 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;
@@ -67,13 +67,9 @@ use std::path::Path;
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::mpsc::Sender;
-use tracing::instrument;
+use tracing::*;
 use url::Url;
 
-#[cfg(target_arch = "wasm32")]
-#[cfg(debug_assertions)]
-use wasm_utils::console_log;
-
 #[cfg(all(
     not(target_arch = "wasm32"),
     feature = "fs-surb-storage",
@@ -84,28 +80,10 @@ pub mod non_wasm_helpers;
 pub mod helpers;
 pub mod storage;
 
-#[derive(Clone, Copy, Debug)]
-pub enum MixnetClientEvent {
-    Traffic(MixTrafficEvent),
-}
-
-pub type EventReceiver = mpsc::UnboundedReceiver<MixnetClientEvent>;
-#[derive(Clone)]
-pub struct EventSender(pub mpsc::UnboundedSender<MixnetClientEvent>);
-
-impl EventSender {
-    pub fn send(&self, event: MixnetClientEvent) {
-        if let Err(err) = self.0.unbounded_send(event) {
-            tracing::warn!("Failed to send error event. The caller event reader was closed: {err}");
-        }
-    }
-}
-
 #[derive(Clone)]
 pub struct ClientInput {
     pub connection_command_sender: ConnectionCommandSender,
     pub input_sender: InputMessageSender,
-    pub client_request_sender: ClientRequestSender,
 }
 
 impl ClientInput {
@@ -117,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> {
@@ -217,8 +195,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>,
-    event_tx: Option<EventSender>,
+    shutdown: Option<TaskClient>,
     user_agent: Option<UserAgent>,
 
     setup_method: GatewaySetup,
@@ -247,7 +224,6 @@ where
             custom_topology_provider: None,
             custom_gateway_transceiver: None,
             shutdown: None,
-            event_tx: None,
             user_agent: None,
             setup_method: GatewaySetup::MustLoad { gateway_id: None },
             #[cfg(unix)]
@@ -305,17 +281,11 @@ 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
     }
 
-    #[must_use]
-    pub fn with_event_tx(mut self, event_tx: EventSender) -> Self {
-        self.event_tx = Some(event_tx);
-        self
-    }
-
     #[must_use]
     pub fn with_user_agent(mut self, user_agent: UserAgent) -> Self {
         self.user_agent = Some(user_agent);
@@ -346,18 +316,6 @@ where
         details.client_address()
     }
 
-    fn start_event_control(
-        parent_event_tx: Option<EventSender>,
-        children_event_rx: EventReceiver,
-        shutdown_tracker: &ShutdownTracker,
-    ) {
-        let event_control = EventControl::new(parent_event_tx, children_event_rx);
-        shutdown_tracker.try_spawn_named_with_shutdown(
-            async move { event_control.run().await },
-            "EventControl",
-        );
-    }
-
     // future constantly pumping loop cover traffic at some specified average rate
     // the pumped traffic goes to the MixTrafficController
     fn start_cover_traffic_stream(
@@ -367,11 +325,11 @@ where
         topology_accessor: TopologyAccessor,
         mix_tx: BatchMixMessageSender,
         stats_tx: ClientStatsSender,
-        shutdown_tracker: &ShutdownTracker,
+        task_client: TaskClient,
     ) {
-        tracing::info!("Starting loop cover traffic stream...");
+        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,
@@ -380,13 +338,13 @@ 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)]
-    #[instrument(skip_all)]
     fn start_real_traffic_controller(
         controller_config: real_messages_control::Config,
         key_rotation_config: KeyRotationConfig,
@@ -399,12 +357,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,
     ) {
-        tracing::info!("Starting real traffic stream...");
+        info!("Starting real traffic stream...");
 
-        let real_messages_controller = RealMessagesController::new(
+        RealMessagesController::new(
             controller_config,
             key_rotation_config,
             ack_receiver,
@@ -417,101 +376,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,
     ) {
-        tracing::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",
-        );
+        info!("Starting received messages buffer controller...");
+        let controller: ReceivedMessagesBufferController<SphinxMessageReceiver> =
+            ReceivedMessagesBufferController::new(
+                local_encryption_keypair,
+                query_receiver,
+                mixnet_receiver,
+                reply_key_storage,
+                reply_controller_sender,
+                metrics_reporter,
+                shutdown,
+            );
+        controller.start()
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[instrument(skip_all)]
     async fn start_gateway_client(
         config: &Config,
         initialisation_result: InitialisationResult,
@@ -520,7 +415,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,
@@ -539,7 +434,7 @@ where
                     packet_router,
                     bandwidth_controller,
                     stats_reporter,
-                    shutdown_tracker.clone_shutdown_token(),
+                    shutdown,
                 )
             } else {
                 let cfg = GatewayConfig::new(
@@ -564,7 +459,7 @@ where
                     stats_reporter,
                     #[cfg(unix)]
                     connection_fd_callback,
-                    shutdown_tracker.clone_shutdown_token(),
+                    shutdown,
                 )
             };
 
@@ -600,7 +495,7 @@ where
             details_store
                 .upgrade_stored_remote_gateway_key(gateway_client.gateway_identity(), &updated_key)
                 .await.map_err(|err| {
-                tracing::error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
+                error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
                 ClientCoreError::GatewaysDetailsStoreError { source: Box::new(err) }
             })?
         }
@@ -618,7 +513,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,
@@ -628,7 +522,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,
@@ -645,6 +539,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)
             };
@@ -660,7 +555,7 @@ where
             stats_reporter,
             #[cfg(unix)]
             connection_fd_callback,
-            shutdown_tracker,
+            shutdown,
         )
         .await?;
 
@@ -691,24 +586,26 @@ 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
-            tracing::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
-        tracing::info!("Obtaining initial network topology");
+        info!("Obtaining initial network topology");
         topology_refresher.try_refresh().await;
 
         if let Err(err) = topology_refresher.ensure_topology_is_routable().await {
@@ -734,13 +631,13 @@ where
                     .wait_for_gateway(local_gateway, waiting_timeout)
                     .await
                 {
-                    tracing::error!(
+                    error!(
                         "the gateway did not come back online within the specified timeout: {err}"
                     );
                     return Err(err.into());
                 }
             } else {
-                tracing::error!("the gateway we're supposedly connected to does not exist. We'll not be able to send any packets to ourselves: {err}");
+                error!("the gateway we're supposedly connected to does not exist. We'll not be able to send any packets to ourselves: {err}");
                 return Err(err.into());
             }
         }
@@ -748,11 +645,8 @@ where
         if !topology_config.disable_refreshing {
             // don't spawn the refresher if we don't want to be refreshing the topology.
             // only use the initial values obtained
-            tracing::info!("Starting topology refresher...");
-            shutdown_tracker.try_spawn_named_with_shutdown(
-                async move { topology_refresher.run().await },
-                "TopologyRefresher",
-            );
+            info!("Starting topology refresher...");
+            topology_refresher.start();
         }
 
         Ok(())
@@ -763,9 +657,9 @@ where
         user_agent: Option<UserAgent>,
         client_stats_id: String,
         input_sender: Sender<InputMessage>,
-        shutdown_tracker: &ShutdownTracker,
+        task_client: TaskClient,
     ) -> ClientStatsSender {
-        tracing::info!("Starting statistics control...");
+        info!("Starting statistics control...");
         StatisticsControl::create_and_start(
             config.debug.stats_reporting,
             user_agent
@@ -773,30 +667,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,
-        event_tx: EventSender,
+        shutdown: TaskClient,
     ) -> (BatchMixMessageSender, ClientRequestSender) {
-        tracing::info!("Starting mix traffic controller...");
-        let mut mix_traffic_controller = MixTrafficController::new(
-            gateway_transceiver,
-            shutdown_tracker.clone_shutdown_token(),
-            event_tx,
-        );
-
-        let mix_tx = mix_traffic_controller.mix_rx();
-        let client_tx = mix_traffic_controller.client_tx();
-
-        shutdown_tracker.try_spawn_named(
-            async move { mix_traffic_controller.run().await },
-            "MixTrafficController",
-        );
-
+        info!("Starting mix traffic controller...");
+        let (mix_traffic_controller, mix_tx, client_tx) =
+            MixTrafficController::new(gateway_transceiver, shutdown);
+        mix_traffic_controller.start();
         (mix_tx, client_tx)
     }
 
@@ -804,7 +686,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,
@@ -829,14 +711,13 @@ where
             })?;
 
         let store_clone = mem_store.clone();
-        let shutdown_token = shutdown_tracker.clone_shutdown_token();
-        shutdown_tracker.try_spawn_named(
+        spawn_future!(
             async move {
                 persistent_storage
-                    .flush_on_shutdown(store_clone, shutdown_token)
+                    .flush_on_shutdown(store_clone, shutdown)
                     .await
             },
-            "PersistentReplyStorage::flush_on_shutdown",
+            "PersistentReplyStorage::flush_on_shutdown"
         );
 
         Ok(mem_store)
@@ -854,7 +735,7 @@ where
     {
         // if client keys do not exist already, create and persist them
         if key_store.load_keys().await.is_err() {
-            tracing::info!("could not find valid client keys - a new set will be generated");
+            info!("could not find valid client keys - a new set will be generated");
             let mut rng = OsRng;
             let keys = if let Some(derivation_material) = derivation_material {
                 ClientKeys::from_master_key(&mut rng, &derivation_material)
@@ -875,8 +756,15 @@ where
         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)?;
+        let mut builder = nym_http_api_client::Client::builder::<
+            _,
+            nym_validator_client::models::RequestError,
+        >(nym_api_urls[0].clone())
+        .map_err(|e| ClientCoreError::NymApiQueryFailure {
+            source: nym_validator_client::nym_api::error::NymAPIError::GenericRequestFailure(
+                e.to_string(),
+            ),
+        })?;
 
         if let Some(user_agent) = user_agent {
             builder = builder.with_user_agent(user_agent);
@@ -884,7 +772,13 @@ where
 
         builder = builder.with_bincode();
 
-        builder.build().map_err(ClientCoreError::from)
+        builder
+            .build::<nym_validator_client::models::RequestError>()
+            .map_err(|e| ClientCoreError::NymApiQueryFailure {
+                source: nym_validator_client::nym_api::error::NymAPIError::GenericRequestFailure(
+                    e.to_string(),
+                ),
+            })
     }
 
     async fn determine_key_rotation_state(
@@ -893,7 +787,6 @@ where
         Ok(client.get_key_rotation_info().await?.into())
     }
 
-    #[instrument(skip_all)]
     pub async fn start_base(mut self) -> Result<BaseClient, ClientCoreError>
     where
         S::ReplyStore: Send + Sync,
@@ -902,12 +795,7 @@ where
         <S::CredentialStore as CredentialStorage>::StorageError: Send + Sync + 'static,
         <S::GatewaysDetailsStore as GatewaysDetailsStore>::StorageError: Sync + Send,
     {
-        tracing::info!("Starting nym client");
-        #[cfg(debug_assertions)]
-        #[cfg(target_arch = "wasm32")]
-        {
-            console_log!("Starting base Nym Client");
-        }
+        info!("Starting nym client");
 
         // derive (or load) client keys and gateway configuration
         let init_res = Self::initialise_keys_and_gateway(
@@ -936,22 +824,17 @@ where
         // channels responsible for controlling real messages
         let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
 
-        // channels responsible for event management
-        let (event_sender, event_receiver) = mpsc::unbounded();
-
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
         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()?,
-        };
-
-        Self::start_event_control(self.event_tx, event_receiver, &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) =
@@ -983,7 +866,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
@@ -993,14 +876,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(
@@ -1013,7 +896,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();
@@ -1021,7 +904,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?;
 
@@ -1031,8 +914,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
@@ -1042,8 +925,7 @@ where
 
         let (message_sender, client_request_sender) = Self::start_mix_traffic_controller(
             gateway_transceiver,
-            &shutdown_tracker.child_tracker(),
-            EventSender(event_sender),
+            shutdown.fork("mix_traffic_controller"),
         );
 
         // Channels that the websocket listener can use to signal downstream to the real traffic
@@ -1072,8 +954,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
@@ -1089,19 +972,12 @@ where
                 shared_topology_accessor.clone(),
                 message_sender,
                 stats_reporter.clone(),
-                &shutdown_tracker.child_tracker(),
+                shutdown.fork("cover_traffic_stream"),
             );
         }
 
-        tracing::debug!("Core client startup finished!");
-        tracing::debug!("The address of this client is: {self_address}");
-
-        #[cfg(debug_assertions)]
-        #[cfg(target_arch = "wasm32")]
-        {
-            console_log!("Core client startup finished!");
-            console_log!("Rust::start_base: the address of this client is: {self_address}");
-        }
+        debug!("Core client startup finished!");
+        debug!("The address of this client is: {self_address}");
 
         Ok(BaseClient {
             address: self_address,
@@ -1110,7 +986,6 @@ where
                 client_input: ClientInput {
                     connection_command_sender: client_connection_tx,
                     input_sender,
-                    client_request_sender,
                 },
             },
             client_output: ClientOutputStatus::AwaitingConsumer {
@@ -1125,7 +1000,8 @@ where
                 gateway_connection: GatewayConnection { gateway_ws_fd },
             },
             stats_reporter,
-            shutdown_handle: 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,
         })
@@ -1139,7 +1015,8 @@ pub struct BaseClient {
     pub client_output: ClientOutputStatus,
     pub client_state: ClientState,
     pub stats_reporter: ClientStatsSender,
-    pub shutdown_handle: ShutdownTracker,
+    pub client_request_sender: ClientRequestSender,
+    pub task_handle: TaskHandle,
     pub forget_me: ForgetMe,
     pub remember_me: RememberMe,
 }

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

@@ -3,7 +3,7 @@
 
 use crate::client::mix_traffic::BatchMixMessageSender;
 use crate::client::topology_control::TopologyAccessor;
-use crate::config;
+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,
         }
     }
 
@@ -225,24 +230,17 @@ impl LoopCoverTrafficStream<OsRng> {
         // JS: due to identical logical structure to OutQueueControl::on_message(), this is also
         // presumably required to prevent bugs in the future. Exact reason is still unknown to me.
 
+        // TODO: temporary and BAD workaround for wasm (we should find a way to yield here in wasm)
         #[cfg(not(target_arch = "wasm32"))]
-        {
-            tokio::task::yield_now().await;
-        }
-
-        #[cfg(target_arch = "wasm32")]
-        {
-            tokio_with_wasm::task::yield_now().await;
-        }
+        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
@@ -252,11 +250,32 @@ 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");
+            },
+            "LoopCoverTrafficStream"
+        )
     }
 }

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

@@ -1,40 +0,0 @@
-// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use futures::StreamExt;
-
-use crate::client::base_client::{EventReceiver, EventSender, MixnetClientEvent};
-
-/// Launches and manages task events, propagating upwards what is not strictly internal.
-pub(crate) struct EventControl {
-    parent_event_tx: Option<EventSender>,
-    children_event_rx: EventReceiver,
-}
-
-impl EventControl {
-    pub(crate) fn new(
-        parent_event_tx: Option<EventSender>,
-        children_event_rx: EventReceiver,
-    ) -> Self {
-        EventControl {
-            parent_event_tx,
-            children_event_rx,
-        }
-    }
-
-    fn is_internal(event: MixnetClientEvent) -> bool {
-        match event {
-            MixnetClientEvent::Traffic(_) => false,
-        }
-    }
-
-    pub(crate) async fn run(mut self) {
-        while let Some(event) = self.children_event_rx.next().await {
-            if let Some(parent_event_tx) = &self.parent_event_tx {
-                if !Self::is_internal(event) {
-                    parent_event_tx.send(event);
-                }
-            }
-        }
-    }
-}

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

@@ -2,9 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #![allow(unused_imports)]
+
 use std::time::Duration;
 
-#[cfg(target_arch = "wasm32")]
 pub use wasmtimer::{std::Instant, tokio::*};
 pub type IntervalStream = gloo_timers::future::IntervalStream;
 

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

@@ -29,8 +29,6 @@ pub enum InputMessage {
         data: Vec<u8>,
         lane: TransmissionLane,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     },
 
     /// Creates a message used for a duplex anonymous communication where the recipient
@@ -47,8 +45,6 @@ pub enum InputMessage {
         reply_surbs: u32,
         lane: TransmissionLane,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     },
 
     /// Attempt to use our internally received and stored `ReplySurb` to send the message back
@@ -94,16 +90,12 @@ impl InputMessage {
         data: Vec<u8>,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     ) -> Self {
         let message = InputMessage::Regular {
             recipient,
             data,
             lane,
             max_retransmissions: None,
-            #[cfg(feature = "otel")]
-            trace_id,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -118,8 +110,6 @@ impl InputMessage {
         reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     ) -> Self {
         let message = InputMessage::Anonymous {
             recipient,
@@ -127,8 +117,6 @@ impl InputMessage {
             reply_surbs,
             lane,
             max_retransmissions: None,
-            #[cfg(feature = "otel")]
-            trace_id,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -197,14 +185,4 @@ impl InputMessage {
         self.set_max_retransmissions(max_retransmissions);
         self
     }
-
-    #[cfg(feature = "otel")]
-    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

@@ -1,13 +1,12 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::client::{
-    base_client::{EventSender, MixnetClientEvent},
-    mix_traffic::transceiver::GatewayTransceiver,
-};
+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;
 
@@ -25,33 +24,28 @@ const MAX_FAILURE_COUNT: usize = 100;
 // that's also disgusting.
 pub struct Empty;
 
-#[derive(Clone, Copy, Debug)]
-pub enum MixTrafficEvent {
-    FailedSendingSphinx,
-}
-
 pub struct MixTrafficController {
     gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
 
-    mix_tx: BatchMixMessageSender,
     mix_rx: BatchMixMessageReceiver,
     client_rx: ClientRequestReceiver,
-    client_tx: ClientRequestSender,
 
     // TODO: this is temporary work-around.
     // in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
     consecutive_gateway_failure_count: usize,
 
-    shutdown_token: ShutdownToken,
-    event_tx: EventSender,
+    task_client: TaskClient,
 }
 
 impl MixTrafficController {
     pub fn new<T>(
         gateway_transceiver: T,
-        shutdown_token: ShutdownToken,
-        event_tx: EventSender,
-    ) -> MixTrafficController
+        task_client: TaskClient,
+    ) -> (
+        MixTrafficController,
+        BatchMixMessageSender,
+        ClientRequestSender,
+    )
     where
         T: GatewayTransceiver + Send + 'static,
     {
@@ -60,32 +54,41 @@ impl MixTrafficController {
 
         let (client_sender, client_receiver) = tokio::sync::mpsc::channel(8);
 
-        MixTrafficController {
-            gateway_transceiver: Box::new(gateway_transceiver),
-            mix_tx: message_sender,
-            mix_rx: message_receiver,
-            client_rx: client_receiver,
-            client_tx: client_sender,
-            consecutive_gateway_failure_count: 0,
-            shutdown_token,
-            event_tx,
-        }
+        (
+            MixTrafficController {
+                gateway_transceiver: Box::new(gateway_transceiver),
+                mix_rx: message_receiver,
+                client_rx: client_receiver,
+                consecutive_gateway_failure_count: 0,
+                task_client,
+            },
+            message_sender,
+            client_sender,
+        )
     }
 
     pub fn new_dynamic(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
-        shutdown_token: ShutdownToken,
-        event_tx: EventSender,
-    ) -> MixTrafficController {
-        Self::new(gateway_transceiver, shutdown_token, event_tx)
-    }
-
-    pub fn client_tx(&self) -> ClientRequestSender {
-        self.client_tx.clone()
-    }
-
-    pub fn mix_rx(&self) -> BatchMixMessageSender {
-        self.mix_tx.clone()
+        task_client: TaskClient,
+    ) -> (
+        MixTrafficController,
+        BatchMixMessageSender,
+        ClientRequestSender,
+    ) {
+        let (message_sender, message_receiver) =
+            tokio::sync::mpsc::channel(MIX_MESSAGE_RECEIVER_BUFFER_SIZE);
+        let (client_sender, client_receiver) = tokio::sync::mpsc::channel(8);
+        (
+            MixTrafficController {
+                gateway_transceiver,
+                mix_rx: message_receiver,
+                client_rx: client_receiver,
+                consecutive_gateway_failure_count: 0,
+                task_client,
+            },
+            message_sender,
+            client_sender,
+        )
     }
 
     async fn on_messages(
@@ -104,7 +107,7 @@ impl MixTrafficController {
 
         tokio::select! {
             biased;
-            _ = self.shutdown_token.cancelled() => {
+            _ = self.task_client.recv() => {
                 trace!("received shutdown while handling messages");
                 Ok(())
             }
@@ -124,7 +127,7 @@ impl MixTrafficController {
     async fn on_client_request(&mut self, client_request: ClientRequest) {
         tokio::select! {
             biased;
-             _ = self.shutdown_token.cancelled() => {
+             _ = self.task_client.recv() => {
                 trace!("received shutdown while handling client request");
             }
             result = self.gateway_transceiver.send_client_request(client_request) => {
@@ -135,37 +138,51 @@ impl MixTrafficController {
         }
     }
 
-    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_rx should never error out as we're holding one instance of the sender
-
-                Some(mix_packets) = self.mix_rx.recv() =>  {
-                    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.event_tx.send(MixnetClientEvent::Traffic(MixTrafficEvent::FailedSendingSphinx));
+    pub fn start(mut self) {
+        spawn_future!(
+            async move {
+                debug!("Started MixTrafficController with graceful shutdown support");
+                while !self.task_client.is_shutdown() {
+                    tokio::select! {
+                        biased;
+                        _ = self.task_client.recv() => {
+                            tracing::trace!("MixTrafficController: Received shutdown");
                             break;
                         }
+                        mix_packets = self.mix_rx.recv() => match mix_packets {
+                            Some(mix_packets) => {
+                                if let Err(err) = self.on_messages(mix_packets).await {
+                                    error!("Failed to send sphinx packet(s) to the gateway: {err}");
+                                    if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
+                                        // Disconnect from the gateway. If we should try to re-connect
+                                        // is handled at a higher layer.
+                                        error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
+                                        // Do we need to handle the embedded mixnet client case
+                                        // separately?
+                                        self.task_client.send_we_stopped(Box::new(ClientCoreError::GatewayFailedToForwardMessages));
+                                        break;
+                                    }
+                                }
+                            },
+                            None => {
+                                tracing::trace!("MixTrafficController: Stopping since channel closed");
+                                break;
+                            }
+                        },
+                        client_request = self.client_rx.recv() => match client_request {
+                            Some(client_request) => {
+                                self.on_client_request(client_request).await;
+                            },
+                            None => {
+                                tracing::trace!("MixTrafficController, client request channel closed");
+                            }
+                        },
                     }
-                },
-                // client_rx should never error out as we're holding one instance of the sender
-                Some(client_request) = self.client_rx.recv() => {
-                    self.on_client_request(client_request).await;
                 }
-            }
-        }
-        debug!("MixTrafficController: Exiting");
+                self.task_client.recv_timeout().await;
+                tracing::debug!("MixTrafficController: Exiting");
+            },
+            "MixTrafficController"
+        );
     }
 }

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

@@ -3,7 +3,6 @@
 
 pub mod base_client;
 pub mod cover_traffic_stream;
-pub(crate) mod event_control;
 pub(crate) mod helpers;
 pub mod inbound_messages;
 pub mod key_manager;

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,
         }
     }
 
@@ -222,9 +226,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 +251,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 +272,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,27 +85,16 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     ) {
         if let Err(err) = self
             .message_handler
-            .try_send_plain_message(
-                recipient,
-                content,
-                lane,
-                packet_type,
-                max_retransmissions,
-                #[cfg(feature = "otel")]
-                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,
@@ -107,8 +103,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     ) {
         if let Err(err) = self
             .message_handler
@@ -119,8 +113,6 @@ where
                 lane,
                 packet_type,
                 max_retransmissions,
-                #[cfg(feature = "otel")]
-                trace_id,
             )
             .await
         {
@@ -129,36 +121,20 @@ where
     }
 
     #[allow(clippy::panic)]
-    #[instrument(skip_all)]
     async fn on_input_message(&mut self, msg: InputMessage) {
-        #[cfg(feature = "otel")]
-        let trace_id = msg.trace_id();
-        #[cfg(feature = "otel")]
-        if let Some(tid) = trace_id {
-            tracing::info!("Processing input message with trace_id: {:?}", tid);
-        }
-
         match msg {
             InputMessage::Regular {
                 recipient,
                 data,
                 lane,
                 max_retransmissions,
-                ..
             } => {
-                #[cfg(feature = "otel")]
-                info!(
-                    "Handling regular input message with trace_id: {:?}",
-                    trace_id
-                );
                 self.handle_plain_message(
                     recipient,
                     data,
                     lane,
                     PacketType::Mix,
                     max_retransmissions,
-                    #[cfg(feature = "otel")]
-                    trace_id,
                 )
                 .await
             }
@@ -168,13 +144,7 @@ where
                 reply_surbs,
                 lane,
                 max_retransmissions,
-                ..
             } => {
-                #[cfg(feature = "otel")]
-                tracing::info!(
-                    "Handling anonymous input message with trace_id: {:?}",
-                    trace_id
-                );
                 self.handle_repliable_message(
                     recipient,
                     data,
@@ -182,8 +152,6 @@ where
                     lane,
                     PacketType::Mix,
                     max_retransmissions,
-                    #[cfg(feature = "otel")]
-                    trace_id,
                 )
                 .await
             }
@@ -193,8 +161,6 @@ where
                 lane,
                 max_retransmissions,
             } => {
-                #[cfg(feature = "otel")]
-                info!("Handling reply input message with trace_id: {:?}", trace_id);
                 self.handle_reply(recipient_tag, data, lane, max_retransmissions)
                     .await;
             }
@@ -208,21 +174,13 @@ where
                     data,
                     lane,
                     max_retransmissions,
-                    ..
                 } => {
-                    #[cfg(feature = "otel")]
-                    tracing::info!(
-                        "Handling regular input message with trace_id: {:?}",
-                        trace_id
-                    );
                     self.handle_plain_message(
                         recipient,
                         data,
                         lane,
                         packet_type,
                         max_retransmissions,
-                        #[cfg(feature = "otel")]
-                        trace_id,
                     )
                     .await
                 }
@@ -232,7 +190,6 @@ where
                     reply_surbs,
                     lane,
                     max_retransmissions,
-                    ..
                 } => {
                     self.handle_repliable_message(
                         recipient,
@@ -241,8 +198,6 @@ where
                         lane,
                         packet_type,
                         max_retransmissions,
-                        #[cfg(feature = "otel")]
-                        trace_id,
                     )
                     .await
                 }
@@ -266,14 +221,13 @@ where
         };
     }
 
-    #[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() => {
+                _ = self.task_client.recv() => {
                     tracing::trace!("InputMessageListener: Received shutdown");
                     break;
                 }
@@ -289,6 +243,7 @@ where
 
             }
         }
+        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,51 @@ 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!");
+            },
+            "AcknowledgementController::AcknowledgementListener"
+        );
+
+        spawn_future!(
+            async move {
+                input_message_listener.run().await;
+                debug!("The input listener has finished execution!");
+            },
+            "AcknowledgementController::InputMessageListener"
+        );
+
+        spawn_future!(
+            async move {
+                retransmission_request_listener.run(packet_type).await;
+                debug!("The retransmission request listener has finished execution!");
+            },
+            "AcknowledgementController::RetransmissionRequestListener"
+        );
+
+        spawn_future!(
+            async move {
+                sent_notification_listener.run().await;
+                debug!("The sent notification listener has finished execution!");
+            },
+            "AcknowledgementController::SentNotificationListener"
+        );
+
+        spawn_future!(
+            async move {
+                action_controller.run().await;
+                debug!("The controller has finished execution!");
+            },
+            "AcknowledgementController::ActionController"
+        );
     }
 }

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,19 +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,
-                #[cfg(feature = "otel")]
-                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,
@@ -102,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
             }
@@ -154,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
@@ -170,18 +174,18 @@ 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() => {
+                 _ = self.task_client.recv() => {
                     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;
@@ -190,6 +194,7 @@ where
 
             }
         }
+        self.task_client.recv_timeout().await;
         tracing::debug!("RetransmissionRequestListener: Exiting");
     }
 }

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

@@ -5,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)]
@@ -189,7 +189,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 +205,7 @@ where
         topology_access: TopologyAccessor,
         reply_key_storage: SentReplyKeys,
         tag_storage: UsedSenderTags,
-        shutdown_token: ShutdownToken,
+        task_client: TaskClient,
     ) -> Self
     where
         R: Copy,
@@ -228,7 +228,7 @@ where
             topology_access,
             reply_key_storage,
             tag_storage,
-            shutdown_token,
+            task_client,
         }
     }
 
@@ -476,7 +476,6 @@ where
         self.forward_messages(msgs, lane).await;
     }
 
-    #[instrument(skip_all)]
     pub(crate) async fn try_send_plain_message(
         &mut self,
         recipient: Recipient,
@@ -484,8 +483,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        trace_id: Option<[u8; 12]>,
     ) -> Result<(), PreparationError> {
         let message = NymMessage::new_plain(message);
         self.try_split_and_send_non_reply_message(
@@ -494,13 +491,10 @@ where
             lane,
             packet_type,
             max_retransmissions,
-            #[cfg(feature = "otel")]
-            trace_id,
         )
         .await
     }
 
-    #[instrument(skip_all)]
     pub(crate) async fn try_split_and_send_non_reply_message(
         &mut self,
         message: NymMessage,
@@ -508,8 +502,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        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
@@ -542,8 +534,6 @@ where
                 &self.config.ack_key,
                 &recipient,
                 packet_type,
-                #[cfg(feature = "otel")]
-                trace_id,
             )?;
 
             let real_message = RealMessage::new(
@@ -595,8 +585,6 @@ where
             TransmissionLane::AdditionalReplySurbs,
             packet_type,
             max_retransmissions,
-            #[cfg(feature = "otel")]
-            None,
         )
         .await?;
 
@@ -614,8 +602,6 @@ where
         lane: TransmissionLane,
         packet_type: PacketType,
         max_retransmissions: Option<u32>,
-        #[cfg(feature = "otel")]
-        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);
@@ -639,8 +625,6 @@ where
             lane,
             packet_type,
             max_retransmissions,
-            #[cfg(feature = "otel")]
-            trace_id,
         )
         .await?;
 
@@ -655,8 +639,6 @@ where
         recipient: Recipient,
         chunk: Fragment,
         packet_type: PacketType,
-        #[cfg(feature = "otel")]
-        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;
@@ -668,8 +650,6 @@ where
             &self.config.ack_key,
             &recipient,
             packet_type,
-            #[cfg(feature = "otel")]
-            trace_id,
         )?;
 
         Ok(prepared_fragment)
@@ -732,7 +712,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}");
             }
         }
@@ -743,7 +723,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}");
             }
         }
@@ -757,7 +737,7 @@ where
     ) {
         tokio::select! {
             biased;
-            _ = self.shutdown_token.cancelled() => {
+            _ = self.task_client.recv() => {
                 trace!("received shutdown while attempting to forward mixnet messages");
             }
             sending_res = self.real_message_sender.send((messages, transmission_lane)) => {

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

@@ -14,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,26 @@ 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!");
+            },
+            "RealMessagesController::OutQueueControl)"
+        );
+        spawn_future!(
+            async move {
+                reply_control.run().await;
+                debug!("The reply controller has finished execution!");
+            },
+            "RealMessagesController::ReplyController"
+        );
+
+        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;
@@ -31,9 +31,9 @@ use tracing::*;
 #[cfg(not(target_arch = "wasm32"))]
 use tokio::time::{sleep, Sleep};
 
-// use wasm_utils::console_log;
 #[cfg(target_arch = "wasm32")]
 use wasmtimer::tokio::{sleep, Sleep};
+
 mod sending_delay_controller;
 
 /// Configurable parameters of the `OutQueueControl`
@@ -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,
         }
     }
 
@@ -282,7 +282,7 @@ where
 
         let sending_res = tokio::select! {
             biased;
-            _ = self.shutdown_token.cancelled() => {
+            _ = self.task_client.recv() => {
                 trace!("received shutdown signal while attempting to send mix message");
                 return
             }
@@ -293,7 +293,7 @@ where
 
         match sending_res {
             Err(_) => {
-                if !self.shutdown_token.is_cancelled() {
+                if !self.task_client.is_shutdown_poll() {
                     tracing::error!(
                         "failed to send mixnet packet due to closed channel (outside of shutdown!)"
                     );
@@ -325,15 +325,9 @@ where
         // ready and hence was immediately re-scheduled causing other tasks to be starved;
         // yield makes it go back the scheduling queue regardless of its value availability
 
+        // TODO: temporary and BAD workaround for wasm (we should find a way to yield here in wasm)
         #[cfg(not(target_arch = "wasm32"))]
-        {
-            tokio::task::yield_now().await;
-        }
-
-        #[cfg(target_arch = "wasm32")]
-        {
-            tokio_with_wasm::task::yield_now().await;
-        }
+        tokio::task::yield_now().await;
     }
 
     fn on_close_connection(&mut self, connection_id: ConnectionId) {
@@ -542,7 +536,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();
@@ -571,33 +567,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;
@@ -607,16 +602,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,11 +190,11 @@ 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,
         }
     }
 
@@ -315,7 +315,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 +338,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 +465,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 +495,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 +573,48 @@ 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}"),
+                }
+            },
+            "ReceivedMessagesBufferController::FragmentedMessageReceiver"
+        );
+        spawn_future!(
+            async move {
+                request_receiver.run().await;
+            },
+            "ReceivedMessagesBufferController::RequestReceiver"
+        );
     }
 }

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/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,8 +91,6 @@ impl StatisticsControl {
             stats_report.into(),
             TransmissionLane::General,
             None,
-            #[cfg(feature = "otel")]
-            None,
         );
         if let Err(err) = self.report_tx.send(report_message).await {
             tracing::error!("Failed to report client stats: {err:?}");
@@ -90,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"))]
@@ -121,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;
                 },
@@ -149,34 +157,37 @@ 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;
+            },
+            "StatisticsControl"
+        )
+    }
+
     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/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,40 @@ 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");
+            },
+            "TopologyRefresher"
+        )
     }
 }

common/client-core/src/error.rs

@@ -4,7 +4,6 @@
 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;
@@ -57,7 +56,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}"
@@ -243,9 +245,6 @@ pub enum ClientCoreError {
 
     #[error("failed to select valid gateway due to incomputable latency")]
     GatewaySelectionFailure { source: WeightedError },
-
-    #[error("Could not access task registry, {0}")]
-    RegistryAccess(#[from] RegistryAccessError),
 }
 
 impl From<tungstenite::Error> for ClientCoreError {
@@ -256,14 +255,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

@@ -45,7 +45,6 @@ type WsConn = JSWebsocket;
 
 const CONCURRENT_GATEWAYS_MEASURED: usize = 20;
 const MEASUREMENTS: usize = 3;
-const DEFAULT_NYM_API_RETRIES: usize = 3;
 
 #[cfg(not(target_arch = "wasm32"))]
 const CONN_TIMEOUT: Duration = Duration::from_millis(1500);
@@ -133,37 +132,39 @@ impl<'a, G: ConnectableGateway> GatewayWithLatency<'a, G> {
     }
 }
 
-pub async fn gateways_for_init(
+pub async fn gateways_for_init<R: Rng>(
+    rng: &mut R,
     nym_apis: &[Url],
     user_agent: Option<UserAgent>,
     minimum_performance: u8,
     ignore_epoch_roles: bool,
-    retry_count: Option<usize>,
 ) -> Result<Vec<RoutingNode>, ClientCoreError> {
-    // Build client with ALL URLs for fallback support
-    let nym_api_urls: Vec<nym_http_api_client::Url> = nym_apis
-        .iter()
-        .map(|url| nym_http_api_client::Url::from(url.clone()))
-        .collect();
+    let nym_api = nym_apis
+        .choose(rng)
+        .ok_or(ClientCoreError::ListOfNymApisIsEmpty)?;
 
-    if nym_api_urls.is_empty() {
-        return Err(ClientCoreError::ListOfNymApisIsEmpty);
-    }
-
-    let retry_count = retry_count.unwrap_or(DEFAULT_NYM_API_RETRIES);
-    let mut builder = nym_http_api_client::ClientBuilder::new_with_urls(nym_api_urls.clone())
-        .with_retries(retry_count)
-        .with_bincode();
+    // 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::NymAPIError { source: 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 = builder
+        .build::<nym_validator_client::models::RequestError>()
+        .map_err(|e| {
+            ClientCoreError::ValidatorClientError(
+                nym_validator_client::ValidatorClientError::NymAPIError { source: e },
+            )
+        })?;
 
-    tracing::debug!("Fetching list of gateways from: {:?}", nym_api_urls);
+    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)
@@ -175,15 +176,17 @@ pub async fn gateways_for_init(
 
     // filter out gateways below minimum performance and ones that could operate as a mixnode
     // (we don't want instability)
-    let valid_gateways: Vec<RoutingNode> = gateways
+    let valid_gateways = gateways
         .iter()
         .filter(|g| ignore_epoch_roles || !g.supported_roles.mixnode)
         .filter(|g| g.performance.round_to_integer() >= minimum_performance)
         .filter_map(|gateway| gateway.try_into().ok())
-        .collect();
+        .collect::<Vec<_>>();
+    tracing::debug!("After checking validity: {}", valid_gateways.len());
+    tracing::trace!("Valid gateways: {valid_gateways:#?}");
 
     tracing::info!(
-        "Found {} valid gateways after filtering",
+        "and {} after validity and performance filtering",
         valid_gateways.len()
     );
 
@@ -346,20 +349,13 @@ pub(super) fn get_specified_gateway(
     must_use_tls: bool,
 ) -> Result<RoutingNode, ClientCoreError> {
     tracing::debug!("Requesting specified gateway: {gateway_identity}");
-
     let user_gateway = ed25519::PublicKey::from_base58_string(gateway_identity)
         .map_err(ClientCoreError::UnableToCreatePublicKeyFromGatewayId)?;
 
     let gateway = gateways
         .iter()
         .find(|gateway| gateway.identity_key == user_gateway)
-        .ok_or_else(|| {
-            tracing::debug!(
-                "Gateway {gateway_identity} not found in {} available gateways",
-                gateways.len()
-            );
-            ClientCoreError::NoGatewayWithId(gateway_identity.to_string())
-        })?;
+        .ok_or_else(|| ClientCoreError::NoGatewayWithId(gateway_identity.to_string()))?;
 
     let Some(entry_details) = gateway.entry.as_ref() else {
         return Err(ClientCoreError::UnsupportedEntry {
@@ -422,52 +418,3 @@ pub(super) async fn register_with_gateway(
         authenticated_ephemeral_client: gateway_client,
     })
 }
-
-#[cfg(test)]
-mod tests {
-    use url::Url;
-
-    #[test]
-    fn test_single_url_builds_without_retries() {
-        let urls = [Url::parse("https://api.nym.com").unwrap()];
-
-        let nym_api_urls: Vec<nym_http_api_client::Url> = urls
-            .iter()
-            .map(|url| nym_http_api_client::Url::from(url.clone()))
-            .collect();
-
-        assert_eq!(nym_api_urls.len(), 1, "Should have exactly one URL");
-    }
-
-    #[test]
-    fn test_multiple_urls_prepared_for_retries() {
-        let urls = vec![
-            Url::parse("https://api1.nym.com").unwrap(),
-            Url::parse("https://api2.nym.com").unwrap(),
-            Url::parse("https://api3.nym.com").unwrap(),
-        ];
-
-        let nym_api_urls: Vec<nym_http_api_client::Url> = urls
-            .iter()
-            .map(|url| nym_http_api_client::Url::from(url.clone()))
-            .collect();
-
-        assert_eq!(nym_api_urls.len(), 3, "Should have all three URLs");
-        assert!(
-            nym_api_urls.len() > 1,
-            "Multiple URLs trigger retry behavior"
-        );
-    }
-
-    #[test]
-    fn test_empty_url_list_is_detected() {
-        let urls: Vec<Url> = vec![];
-
-        let nym_api_urls: Vec<nym_http_api_client::Url> = urls
-            .iter()
-            .map(|url| nym_http_api_client::Url::from(url.clone()))
-            .collect();
-
-        assert!(nym_api_urls.is_empty(), "Empty list should remain empty");
-    }
-}

common/client-core/src/lib.rs

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

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

@@ -40,7 +40,7 @@ where
     pub async fn flush_on_shutdown(
         mut self,
         mem_state: CombinedReplyStorage,
-        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,
         }
     }
 
@@ -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.

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

@@ -267,7 +267,6 @@ impl Client {
 }
 
 impl SendWithoutResponse for Client {
-    #[instrument(skip(self, packet))]
     fn send_without_response(&self, packet: MixPacket) -> io::Result<()> {
         let address = packet.next_hop_address();
         trace!("Sending packet to {address}");

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

@@ -3,7 +3,7 @@ name = "nym-validator-client"
 version = "0.1.0"
 authors = ["Jędrzej Stuczyński <andrew@nymtech.net>"]
 edition = "2021"
-rust-version = "1.85"
+rust-version = "1.56"
 license.workspace = true
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

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

@@ -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,
 };
@@ -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,
@@ -357,10 +418,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_http_api_client::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 +558,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 +655,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,

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

@@ -24,15 +24,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.base_urls()
         )
     }
 }
@@ -96,10 +91,13 @@ impl TryFrom<ContractVKShare> for EcashApiClient {
         // In non-client applications this resolver can cause warning logs about H2 connection
         // failure. This indicates that the long lived https connection was closed by the remote
         // peer and the resolver will have to reconnect. It should not impact actual functionality
-        let api_client = nym_http_api_client::Client::builder(url_address)
-            .map_err(|e| EcashApiError::ClientError(e.to_string()))?
-            .build()
-            .map_err(|e| EcashApiError::ClientError(e.to_string()))?;
+        let api_client = nym_http_api_client::Client::builder::<
+            _,
+            nym_api_requests::models::RequestError,
+        >(url_address)
+        .map_err(|e| EcashApiError::ClientError(e.to_string()))?
+        .build::<nym_api_requests::models::RequestError>()
+        .map_err(|e| EcashApiError::ClientError(e.to_string()))?;
 
         Ok(EcashApiClient {
             api_client,

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

@@ -89,7 +89,9 @@ 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()) {
+        match nym_http_api_client::Client::builder(url.clone())
+            .and_then(|b| b.build::<nym_api_requests::models::RequestError>())
+        {
             Ok(client) => Some(ClientForConnectionTest::Api(network, url, client)),
             Err(err) => {
                 eprintln!(