Merge branch 'develop-with-release-1.1.0-merged-in' into feature/validator-api-tests

- added missing .git files
- fixed paths
- run the linter

.github/workflows/audit.yml

@@ -16,10 +16,7 @@ jobs:
       - name: Install cargo deny
         run: cargo install --locked cargo-deny
       - name: Run cargo deny
-        run: |
+        run: cargo deny check advisories --hide-inclusion-graph &> .github/workflows/support-files/notifications/deny.message
-          find . -name Cargo.toml -exec cargo deny --manifest-path {} check \
-            advisories -A advisory-not-detected  --hide-inclusion-graph \; &> \
-            >(uniq  &> .github/workflows/support-files/notifications/deny.message )
       - uses: actions/upload-artifact@v3
         with:
           name: report

.github/workflows/build.yml

@@ -29,12 +29,6 @@ jobs:
           override: true
           components: rustfmt, clippy
 
-      - name: Check formatting
-        uses: actions-rs/cargo@v1
-        with:
-          command: fmt
-          args: --all -- --check
-
       - name: Build all binaries
         uses: actions-rs/cargo@v1
         with:
@@ -54,6 +48,12 @@ jobs:
           command: test
           args: --workspace --all-features -- --ignored
 
+      - name: Check formatting
+        uses: actions-rs/cargo@v1
+        with:
+          command: fmt
+          args: --all -- --check
+
       - uses: actions-rs/clippy-check@v1
         name: Clippy checks
         with:
@@ -66,8 +66,6 @@ jobs:
           command: clippy
           args: --workspace -- -D warnings
 
-      # COCONUT stuff
-
       - name: Build all binaries with coconut enabled
         uses: actions-rs/cargo@v1
         with:
@@ -84,4 +82,4 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: clippy
-          args: --all-targets --features=coconut -- -D warnings
+          args: --features=coconut -- -D warnings

.github/workflows/build_on_dispatch.yml

@@ -0,0 +1,72 @@
+name: Continuous integration on dispatch
+
+on: workflow_dispatch
+
+jobs:
+  build:
+    runs-on: [ self-hosted, custom-linux ]
+    # Enable sccache via environment variable
+    env:
+      RUSTC_WRAPPER: /home/ubuntu/.cargo/bin/sccache
+    steps:
+      - name: Install Dependencies (Linux)
+        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev squashfs-tools
+
+      - name: Check out repository code
+        uses: actions/checkout@v2
+
+      - name: Install rust toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+          override: true
+          components: rustfmt, clippy
+
+      - name: Build all binaries
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --workspace
+
+      - name: Run all tests
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --workspace --all-features
+
+      - name: Check formatting
+        uses: actions-rs/cargo@v1
+        with:
+          command: fmt
+          args: --all -- --check
+
+      - uses: actions-rs/clippy-check@v1
+        name: Clippy checks
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          args: --all-features
+
+      - name: Run clippy
+        uses: actions-rs/cargo@v1
+        with:
+          command: clippy
+          args: --workspace -- -D warnings
+
+      - name: Build all binaries with coconut enabled
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --workspace --features=coconut
+
+      - name: Run all tests with coconut enabled
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --workspace --features=coconut
+
+      - name: Run clippy with coconut enabled
+        uses: actions-rs/cargo@v1
+        with:
+          command: clippy
+          args: --features=coconut -- -D warnings

.github/workflows/contracts-build.yml

@@ -1,21 +1,16 @@
 name: Build release of Nym smart contracts
 on:
   workflow_dispatch:
-  release:
+
-    types: [created]
+defaults:
+  run:
+    working-directory: contracts
 
 jobs:
   build:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
-      
-      - name: Check the release tag starts with `nym-contracts-`
-        if: startsWith(github.ref, 'refs/tags/nym-contracts-') == false && github.event_name != 'workflow_dispatch'
-        uses: actions/github-script@v3
-        with:
-          script: |
-            core.setFailed('Release tag did not start with nym-contracts-...')      
 
       - name: Install Rust stable
         uses: actions-rs/toolchain@v1
@@ -26,7 +21,7 @@ jobs:
           components: rustfmt, clippy
 
       - name: Build release contracts
-        run: make wasm
+        run: RUSTFLAGS='-C link-arg=-s' cargo build --release --target wasm32-unknown-unknown
 
       - name: Upload Mixnet Contract Artifact
         uses: actions/upload-artifact@v3
@@ -41,11 +36,3 @@ jobs:
           name: vesting_contract.wasm
           path: contracts/target/wasm32-unknown-unknown/release/vesting_contract.wasm
           retention-days: 5
-
-      - name: Upload to release based on tag name
-        uses: softprops/action-gh-release@v1
-        if: github.event_name == 'release'
-        with:
-          files: |
-            contracts/target/wasm32-unknown-unknown/release/vesting_contract.wasm
-            contracts/target/wasm32-unknown-unknown/release/mixnet_contract.wasm

.github/workflows/network-explorer-api.yml

@@ -1,56 +0,0 @@
-name: CI for Network Explorer API
-
-on:
-  workflow_dispatch:
-  release:
-    types: [created]
-    
-env:
-  NETWORK: mainnet    
-
-jobs:
-  publish-nym:
-    strategy:
-      fail-fast: false
-      matrix:
-        platform: [ubuntu-latest]
-
-    runs-on: ${{ matrix.platform }}
-    steps:
-      - uses: actions/checkout@v3
-
-      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
-
-      - name: Check the release tag starts with `nym-explorer-api-`
-        if: startsWith(github.ref, 'refs/tags/nym-explorer-api-') == false && github.event_name != 'workflow_dispatch'
-        uses: actions/github-script@v3
-        with:
-          script: |
-            core.setFailed('Release tag did not start with nym-explorer-api-...')
-
-      - name: Install Rust stable
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: stable
-
-      - name: Build all explorer-api
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
-          args: --manifest-path explorer-api/Cargo.toml --workspace --release
-
-      - name: Upload Artifact
-        uses: actions/upload-artifact@v3
-        with:
-          name: my-artifact
-          path: |
-            target/release/explorer-api
-          retention-days: 30
-
-      - name: Upload to release based on tag name
-        uses: softprops/action-gh-release@v1
-        if: github.event_name == 'release'
-        with:
-          files: |
-            target/release/explorer-api

.github/workflows/nightly_build_matrix_on_dispatch.json

@@ -0,0 +1,50 @@
+[
+  {
+    "os":"ubuntu-latest",
+    "rust":"stable",
+    "runOnEvent":"workflow_dispatch"
+  },
+
+  {
+    "os":"windows-latest",
+    "rust":"stable",
+    "runOnEvent":"workflow_dispatch"
+  },
+  {
+    "os":"macos-latest",
+    "rust":"stable",
+    "runOnEvent":"workflow_dispatch"
+  },
+
+  {
+    "os":"ubuntu-latest",
+    "rust":"beta",
+    "runOnEvent":"workflow_dispatch"
+  },
+  {
+    "os":"windows-latest",
+    "rust":"beta",
+    "runOnEvent":"workflow_dispatch"
+  },
+  {
+    "os":"macos-latest",
+    "rust":"beta",
+    "runOnEvent":"workflow_dispatch"
+  },
+
+  {
+    "os":"ubuntu-latest",
+    "rust":"nightly",
+    "runOnEvent":"workflow_dispatch"
+  },
+  {
+    "os":"windows-latest",
+    "rust":"nightly",
+    "runOnEvent":"workflow_dispatch"
+  },
+  {
+    "os":"macos-latest",
+    "rust":"nightly",
+    "runOnEvent":"workflow_dispatch"
+  }
+]

.github/workflows/nightly_build_on_dispatch.yml

@@ -1,8 +1,6 @@
-name: Nightly builds on latest release
+name: Nightly builds on dispatch
 
-on:
+on: workflow_dispatch
-  schedule:
-    - cron: '14 2 * * *'
 jobs:
   matrix_prep:
     runs-on: ubuntu-latest
@@ -10,40 +8,25 @@ jobs:
       matrix: ${{ steps.set-matrix.outputs.matrix }}
     steps:
       # creates the matrix strategy from nightly_build_matrix_includes.json
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v2
       - id: set-matrix
         uses: JoshuaTheMiller/conditional-build-matrix@main
         with:
-          inputFile: '.github/workflows/nightly_build_matrix_includes.json'
+          inputFile: '.github/workflows/nightly_build_matrix_on_dispatch.json'
           filter: '[?runOnEvent==`${{ github.event_name }}` || runOnEvent==`always`]'
-  get_release:
-    runs-on: ubuntu-latest
-    needs: matrix_prep
-    outputs:
-      output1: ${{ steps.step2.outputs.latest_release }}
-    steps:
-      - name: Check out repository code
-        uses: actions/checkout@v3
-      - name: Fetch all branches
-        run: git fetch --all
-      - name: Set output variable to latest release branch
-        id: step2
-        run: echo "latest_release=$(git branch -r | grep -E 'release/v[0-9]+\.[0-9]+\.[0-9]+' | tail -n 1 | sed 's/  origin\///')" >> $GITHUB_OUTPUT
   build:
-    needs: [get_release,matrix_prep]
+    needs: matrix_prep
     strategy:
       matrix: ${{fromJson(needs.matrix_prep.outputs.matrix)}}
     runs-on: ${{ matrix.os }}
     continue-on-error: ${{ matrix.rust == 'nightly' || matrix.rust == 'beta' || matrix.rust == 'stable' }}
     steps:
       - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
+        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev squashfs-tools
         if: matrix.os == 'ubuntu-latest'
 
-      - name: Check out latest release branch
+      - name: Check out repository code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v2
-        with:
-          ref: ${{needs.get_release.outputs.output1}}
 
       - name: Install rust toolchain
         uses: actions-rs/toolchain@v1
@@ -59,12 +42,6 @@ jobs:
           command: build
           args: --workspace
 
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
       - name: Run all tests
         uses: actions-rs/cargo@v1
         with:
@@ -122,12 +99,6 @@ jobs:
           command: build
           args: --workspace --features=coconut
 
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
       - name: Run all tests with coconut enabled
         uses: actions-rs/cargo@v1
         with:
@@ -174,13 +145,13 @@ jobs:
           args: --manifest-path nym-wallet/Cargo.toml --workspace --all-targets -- -D warnings
 
   notification:
-    needs: [build,get_release]
+    needs: build
     runs-on: ubuntu-latest
     steps:
       - name: Collect jobs status
         uses: technote-space/workflow-conclusion-action@v2
       - name: Check out repository code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v2
       - name: Keybase - Node Install
         if: env.WORKFLOW_CONCLUSION == 'failure'
         run: npm install
@@ -189,14 +160,14 @@ jobs:
         if: env.WORKFLOW_CONCLUSION == 'failure'
         env:
           NYM_NOTIFICATION_KIND: nightly
-          NYM_PROJECT_NAME: "Nym nightly build on latest release"
+          NYM_PROJECT_NAME: "Nym nightly build"
           GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
           GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
-          GIT_BRANCH: "https://github.com/nymtech/nym/tree/${{needs.get_release.outputs.output1}}"
+          GIT_BRANCH: "${GITHUB_REF##*/}"
           KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
           KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
-          KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
+          KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMTECH_TEAM }}"
-          KEYBASE_NYM_CHANNEL: "ci-nightly-release"
+          KEYBASE_NYM_CHANNEL: "${{ secrets.KEYBASE_CHANNEL_DEV_CORE_ID }}"
           IS_SUCCESS: "${{ env.WORKFLOW_CONCLUSION == 'success' }}"
         uses: docker://keybaseio/client:stable-node
         with:

.github/workflows/nightly_build_release2.yml

@@ -1,203 +0,0 @@
-name: Nightly builds on second latest release
-
-on:
-  schedule:
-    - cron: '24 2 * * *'
-jobs:
-  matrix_prep:
-    runs-on: ubuntu-latest
-    outputs:
-      matrix: ${{ steps.set-matrix.outputs.matrix }}
-    steps:
-      # creates the matrix strategy from nightly_build_matrix_includes.json
-      - uses: actions/checkout@v3
-      - id: set-matrix
-        uses: JoshuaTheMiller/conditional-build-matrix@main
-        with:
-          inputFile: '.github/workflows/nightly_build_matrix_includes.json'
-          filter: '[?runOnEvent==`${{ github.event_name }}` || runOnEvent==`always`]'
-  get_release:
-    runs-on: ubuntu-latest
-    needs: matrix_prep
-    outputs:
-      output1: ${{ steps.step2.outputs.latest_release }}
-    steps:
-      - name: Check out repository code
-        uses: actions/checkout@v3
-      - name: Fetch all branches
-        run: git fetch --all
-      - name: Set output variable to latest release branch
-        id: step2
-        run: echo "latest_release=$(git branch -r | grep -E 'release/v[0-9]+\.[0-9]+\.[0-9]+' | tail -n 2 | head -n 1 | sed 's/  origin\///')" >> $GITHUB_OUTPUT
-  build:
-    needs: [get_release,matrix_prep]
-    strategy:
-      matrix: ${{fromJson(needs.matrix_prep.outputs.matrix)}}
-    runs-on: ${{ matrix.os }}
-    continue-on-error: ${{ matrix.rust == 'nightly' || matrix.rust == 'beta' || matrix.rust == 'stable' }}
-    steps:
-      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
-        if: matrix.os == 'ubuntu-latest'
-
-      - name: Check out latest release branch
-        uses: actions/checkout@v3
-        with:
-          ref: ${{needs.get_release.outputs.output1}}
-
-      - name: Install rust toolchain
-        uses: actions-rs/toolchain@v1
-        with:
-          profile: minimal
-          toolchain: ${{ matrix.rust }}
-          override: true
-          components: rustfmt, clippy
-
-      - name: Build all binaries
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
-          args: --workspace
-
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
-      - name: Run all tests
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --workspace
-
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
-      - name: Run expensive tests
-        if: github.ref == 'refs/heads/develop' || github.event.pull_request.base.ref == 'develop' || github.event.pull_request.base.ref == 'master'
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --workspace --all-features -- --ignored
-
-      - name: Check formatting
-        uses: actions-rs/cargo@v1
-        with:
-          command: fmt
-          args: --all -- --check
-
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
-      - uses: actions-rs/clippy-check@v1
-        name: Clippy checks
-        with:
-          token: ${{ secrets.GITHUB_TOKEN }}
-          args: --all-features
-
-      - name: Run clippy
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.rust != 'nightly' }}
-        with:
-          command: clippy
-          args: --workspace --all-targets -- -D warnings
-
-      - name: Reclaim some disk space
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' || matrix.os == 'ubuntu-latest' }}
-        with:
-          command: clean
-
-      # COCONUT stuff
-      - name: Build all binaries with coconut enabled
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
-          args: --workspace --features=coconut
-
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
-      - name: Run all tests with coconut enabled
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --workspace --features=coconut
-
-      - name: Reclaim some disk space (because Windows is being annoying)
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' }}
-        with:
-          command: clean
-
-      - name: Run clippy with coconut enabled
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.rust != 'nightly' }}
-        with:
-          command: clippy
-          args: --workspace --all-targets --features=coconut -- -D warnings
-
-      # nym-wallet (the rust part)
-      - name: Build nym-wallet rust code
-        uses: actions-rs/cargo@v1
-        with:
-          command: build
-          args: --manifest-path nym-wallet/Cargo.toml --workspace
-
-      - name: Run nym-wallet tests
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --manifest-path nym-wallet/Cargo.toml --workspace
-
-      - name: Check nym-wallet formatting
-        uses: actions-rs/cargo@v1
-        with:
-          command: fmt
-          args: --manifest-path nym-wallet/Cargo.toml --all -- --check
-
-      - name: Run clippy for nym-wallet
-        uses: actions-rs/cargo@v1
-        if: ${{ matrix.rust != 'nightly' }}
-        with:
-          command: clippy
-          args: --manifest-path nym-wallet/Cargo.toml --workspace --all-targets -- -D warnings
-
-  notification:
-    needs: [build,get_release]
-    runs-on: ubuntu-latest
-    steps:
-      - name: Collect jobs status
-        uses: technote-space/workflow-conclusion-action@v2
-      - name: Check out repository code
-        uses: actions/checkout@v3
-      - name: Keybase - Node Install
-        if: env.WORKFLOW_CONCLUSION == 'failure'
-        run: npm install
-        working-directory: .github/workflows/support-files
-      - name: Keybase - Send Notification
-        if: env.WORKFLOW_CONCLUSION == 'failure'
-        env:
-          NYM_NOTIFICATION_KIND: nightly
-          NYM_PROJECT_NAME: "Nym nightly build on latest release"
-          GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
-          GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
-          GIT_BRANCH: "https://github.com/nymtech/nym/tree/${{needs.get_release.outputs.output1}}"
-          KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
-          KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
-          KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
-          KEYBASE_NYM_CHANNEL: "ci-nightly-release"
-          IS_SUCCESS: "${{ env.WORKFLOW_CONCLUSION == 'success' }}"
-        uses: docker://keybaseio/client:stable-node
-        with:
-          args: .github/workflows/support-files/notifications/entry_point.sh

.github/workflows/nym-connect.yml

@@ -41,19 +41,19 @@ jobs:
     - name: Keybase - Node Install
       run: npm install
       working-directory: .github/workflows/support-files
-    - name: Keybase - Send Notification
+#    - name: Keybase - Send Notification
-      env:
+#      env:
-        NYM_NOTIFICATION_KIND: nym-connect
+#        NYM_NOTIFICATION_KIND: nym-connect
-        NYM_PROJECT_NAME: "nym-connect"
+#        NYM_PROJECT_NAME: "nym-connect"
-        NYM_CI_WWW_BASE: "${{ secrets.NYM_CI_WWW_BASE }}"
+#        NYM_CI_WWW_BASE: "${{ secrets.NYM_CI_WWW_BASE }}"
-        NYM_CI_WWW_LOCATION: "nym-connect-${{ env.GITHUB_REF_SLUG }}"
+#        NYM_CI_WWW_LOCATION: "nym-connect-${{ env.GITHUB_REF_SLUG }}"
-        GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
+#        GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
-        GIT_BRANCH: "${GITHUB_REF##*/}"
+#        GIT_BRANCH: "${GITHUB_REF##*/}"
-        KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
+#        KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
-        KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
+#        KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
-        KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
+#        KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
-        KEYBASE_NYM_CHANNEL: "ci-nym-connect"
+#        KEYBASE_NYM_CHANNEL: "ci-nym-connect"
-        IS_SUCCESS: "${{ job.status == 'success' }}"
+#        IS_SUCCESS: "${{ job.status == 'success' }}"
-      uses: docker://keybaseio/client:stable-node
+#      uses: docker://keybaseio/client:stable-node
-      with:
+#      with:
-        args: .github/workflows/support-files/notifications/entry_point.sh
+#        args: .github/workflows/support-files/notifications/entry_point.sh

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

@@ -25,9 +25,6 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
-      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
-
       - name: Check the release tag starts with `nym-binaries-`
         if: startsWith(github.ref, 'refs/tags/nym-binaries-') == false && github.event_name != 'workflow_dispatch'
         uses: actions/github-script@v3

.github/workflows/validator-api-tests.yml

@@ -0,0 +1,32 @@
+name: Tests for validator API
+
+on:
+  push:
+    paths:
+      - "validator-api/tests/**"
+
+defaults:
+  run:
+    working-directory: validator-api/tests
+
+jobs:
+  test:
+    name: validator api tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Node v18
+        uses: actions/setup-node@v3
+        with:
+          node-version: 18.1.0
+
+      - name: Install yarn
+        run: yarn install
+
+      - name: Run yarn
+        run: yarn
+
+      - name: Launch tests
+        run: yarn test
+        working-directory: validator-api/tests

CHANGELOG.md

@@ -6,71 +6,35 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ### Added
 
-- binaries: add `-c` shortform for `--config-env-file`
+- nym-cli: added CLI tool for interacting with the Nyx blockchain and Nym mixnet smart contracts ([#1577])
-- websocket-requests: add server response signalling current packet queue length in the client
-- contracts: DKG contract that handles coconut key generation ([#1678][#1708][#1747])
-- validator-api: generate coconut keys interactively, using DKG and multisig contracts ([#1678][#1708][#1747])
-
-### Changed
-
-- clients: add concept of transmission lanes to better handle multiple data streams ([#1720])
-- clients,validator-api: take coconut signers from the chain instead of specifying them via CLI ([#1747])
-- multisig contract: add DKG contract to the list of addresses that can create proposals ([#1747])
-- socks5-client: wait closing inbound connection until data is sent, and throttle incoming data in general ([#1783])
-- nym-cli: improve error reporting/handling and changed `vesting-schedule` queries to use query client instead of signing client
-
-### Fixed
-
-- gateway-client: fix decrypting stored messages on reconnect ([#1786])
-
-### Fixed
-
-- gateway-client: fix decrypting stored messages on reconnect ([#1786])
-- socks5-client: fix shutting down all tasks if anyone of them panics or errors out ([#1805])
-
-[#1678]: https://github.com/nymtech/nym/pull/1678
-[#1708]: https://github.com/nymtech/nym/pull/1708
-[#1720]: https://github.com/nymtech/nym/pull/1720
-[#1747]: https://github.com/nymtech/nym/pull/1747
-[#1783]: https://github.com/nymtech/nym/pull/1783
-[#1786]: https://github.com/nymtech/nym/pull/1786
-[#1805]: https://github.com/nymtech/nym/pull/1805
-
-
-## [v1.1.0](https://github.com/nymtech/nym/tree/v1.1.0) (2022-11-09)
-
-### Added
-
-- clients: add testing-only support for two more extended packet sizes (8kb and 16kb).
-- common/ledger: new library for communicating with a Ledger device ([#1640])
-- native-client/socks5-client/wasm-client: `disable_loop_cover_traffic_stream` Debug config option to disable the separate loop cover traffic stream ([#1666])
-- native-client/socks5-client/wasm-client: `disable_main_poisson_packet_distribution` Debug config option to make the client ignore poisson distribution in the main packet stream and ONLY send real message (and as fast as they come) ([#1664])
-- native-client/socks5-client/wasm-client: `use_extended_packet_size` Debug config option to make the client use 'ExtendedPacketSize' for its traffic (32kB as opposed to 2kB in 1.0.2) ([#1671])
-- network-requester: added additional Blockstream Green wallet endpoint to `example.allowed.list` ([#1611])
-- validator-api: add `interval_operating_cost` and `profit_margin_percent` to compute reward estimation endpoint
 - validator-client: added `query_contract_smart` and `query_contract_raw` on `NymdClient` ([#1558])
+- network-requester: added additional Blockstream Green wallet endpoint to `example.allowed.list` ([#1611](https://github.com/nymtech/nym/pull/1611))
+- common/ledger: new library for communicating with a Ledger device ([#1640])
+- native-client/socks5-client: `disable_loop_cover_traffic_stream` Debug config option to disable the separate loop cover traffic stream ([#1666])
+- native-client/socks5-client: `disable_main_poisson_packet_distribution` Debug config option to make the client ignore poisson distribution in the main packet stream and ONLY send real message (and as fast as they come) ([#1664])
+- native-client/socks5-client: `use_extended_packet_size` Debug config option to make the client use 'ExtendedPacketSize' for its traffic (32kB as opposed to 2kB in 1.0.2) ([#1671])
 - wasm-client: uses updated wasm-compatible `client-core` so that it's now capable of packet retransmission, cover traffic and poisson delay (among other things!) ([#1673])
+- validator-api: add `interval_operating_cost` and `profit_margin_percent` to cmpute reward estimation endpoint
+- native-client/socks5-client/network-requester: improve handling error cases ([#1713])
+- vesting-contract: optional locked token pledge cap per account ([#1687]), defaults to 100_000 NYM
+- clients: add testing-only support for two more extended packet sizes (8kb and 16kb).
 
 ### Fixed
 
+- validator-api, mixnode, gateway should now prefer values in config.toml over mainnet defaults ([#1645])
+- validator-api should now correctly update historical uptimes for all mixnodes and gateways every 24h ([#1721])
 - socks5-client: fix bug where in some cases packet reordering could trigger a connection being closed too early ([#1702],[#1724])
-- validator-api: mixnode, gateway should now prefer values in config.toml over mainnet defaults ([#1645])
-- validator-api: should now correctly update historical uptimes for all mixnodes and gateways every 24h ([#1721])
 
 ### Changed
 
-- clients: bound the sphinx packet channel and reduce sending rate if gateway can't keep up ([#1703],[#1725])
-- gateway-client: will attempt to read now as many as 8 websocket messages at once, assuming they're already available on the socket ([#1669])
-- moved `Percent` struct to `contracts-common`, change affects explorer-api
-- socks5 client: graceful shutdown should fix error on disconnect in nym-connect ([#1591])
-- validator-api: changed error serialization on `inclusion_probability`, `stake-saturation` and `reward-estimation` endpoints to provide more accurate information ([#1681]) 
 - validator-client: made `fee` argument optional for `execute` and `execute_multiple` ([#1541])
+- socks5 client: graceful shutdown should fix error on disconnect in nym-connect ([#1591])
 - wasm-client: fixed build errors on MacOS and changed example JS code to use mainnet ([#1585])
-- validator-api: changes to internal SQL schema due to the mixnet contract revamp ([#1472])
+- gateway-client: will attempt to read now as many as 8 websocket messages at once, assuming they're already available on the socket ([#1669])
-- validator-api: changes to internal data structures due to the mixnet contract revamp ([#1472])
+- validator-api: changed error serialization on `inclusion_probability`, `stake-saturation` and `reward-estimation` endpoints to provide more accurate information ([#1681]) 
-- validator-api: split epoch-operations into multiple separate transactions ([#1472])
+- moved `Percent` struct to to `contracts-common`, change affects explorer-api
+- clients: bound the sphinx packet channel and reduce sending rate if gateway can't keep up ([#1703],[#1725])
 
-[#1472]: https://github.com/nymtech/nym/pull/1472
 [#1541]: https://github.com/nymtech/nym/pull/1541
 [#1558]: https://github.com/nymtech/nym/pull/1558
 [#1577]: https://github.com/nymtech/nym/pull/1577
@@ -78,15 +42,16 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 [#1591]: https://github.com/nymtech/nym/pull/1591
 [#1640]: https://github.com/nymtech/nym/pull/1640
 [#1645]: https://github.com/nymtech/nym/pull/1645
-[#1611]: https://github.com/nymtech/nym/pull/1611
 [#1664]: https://github.com/nymtech/nym/pull/1664
 [#1666]: https://github.com/nymtech/nym/pull/1645
 [#1669]: https://github.com/nymtech/nym/pull/1669
 [#1671]: https://github.com/nymtech/nym/pull/1671
 [#1673]: https://github.com/nymtech/nym/pull/1673
 [#1681]: https://github.com/nymtech/nym/pull/1681
+[#1687]: https://github.com/nymtech/nym/pull/1687
 [#1702]: https://github.com/nymtech/nym/pull/1702
 [#1703]: https://github.com/nymtech/nym/pull/1703
+[#1713]: https://github.com/nymtech/nym/pull/1713
 [#1721]: https://github.com/nymtech/nym/pull/1721
 [#1724]: https://github.com/nymtech/nym/pull/1724
 [#1725]: https://github.com/nymtech/nym/pull/1725

Cargo.lock

@@ -576,19 +576,10 @@ dependencies = [
  "os_str_bytes",
 ]
 
-[[package]]
-name = "client-connections"
-version = "0.1.0"
-dependencies = [
- "futures",
- "log",
-]
-
 [[package]]
 name = "client-core"
-version = "1.1.0"
+version = "1.0.1"
 dependencies = [
- "client-connections",
  "config",
  "crypto",
  "dirs",
@@ -636,18 +627,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "coconut-dkg-common"
-version = "0.1.0"
-dependencies = [
- "contracts-common",
- "cosmwasm-std",
- "cw-utils",
- "multisig-contract-common",
- "schemars",
- "serde",
-]
-
 [[package]]
 name = "coconut-interface"
 version = "0.1.0"
@@ -757,9 +736,7 @@ checksum = "245097e9a4535ee1e3e3931fcfcd55a796a44c643e8596ff6566d68f09b87bbc"
 name = "contracts-common"
 version = "0.1.0"
 dependencies = [
- "bs58",
  "cosmwasm-std",
- "dkg",
  "schemars",
  "serde",
  "serde_json",
@@ -873,9 +850,9 @@ dependencies = [
 
 [[package]]
 name = "cpufeatures"
-version = "0.2.5"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "28d997bd5e24a5928dd43e46dc529867e207907fe0b239c3477d924f7f2ca320"
+checksum = "59a6001667ab124aebae2a495118e11d30984c3a653e99d86d58971708cf5e4b"
 dependencies = [
  "libc",
 ]
@@ -965,6 +942,7 @@ dependencies = [
  "crypto",
  "rand 0.7.3",
  "thiserror",
+ "url",
  "validator-api-requests",
  "validator-client",
 ]
@@ -1243,9 +1221,9 @@ dependencies = [
 
 [[package]]
 name = "cw-utils"
-version = "0.13.4"
+version = "0.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9dbaecb78c8e8abfd6b4258c7f4fbeb5c49a5e45ee4d910d3240ee8e1d714e1b"
+checksum = "babd2c090f39d07ce5bf2556962305e795daa048ce20a93709eb591476e4a29e"
 dependencies = [
  "cosmwasm-std",
  "schemars",
@@ -1253,23 +1231,11 @@ dependencies = [
  "thiserror",
 ]
 
-[[package]]
-name = "cw2"
-version = "0.13.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04cf4639517490dd36b333bbd6c4fbd92e325fd0acf4683b41753bc5eb63bfc1"
-dependencies = [
- "cosmwasm-std",
- "cw-storage-plus",
- "schemars",
- "serde",
-]
-
 [[package]]
 name = "cw3"
-version = "0.13.4"
+version = "0.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fe19462a7f644ba60c19d3443cb90d00c50d9b6b3b0a3a7fca93df8261af979b"
+checksum = "f871854338a54c7bb094d16ffe17212b93b146d9659dbce4c9402a9b77e240ef"
 dependencies = [
  "cosmwasm-std",
  "cw-utils",
@@ -1277,27 +1243,11 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "cw3-fixed-multisig"
-version = "0.13.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df54aa54c13f405ec4ab36b6217538bc957d439eee58f89312db05a79caf6706"
-dependencies = [
- "cosmwasm-std",
- "cw-storage-plus",
- "cw-utils",
- "cw2",
- "cw3",
- "schemars",
- "serde",
- "thiserror",
-]
-
 [[package]]
 name = "cw4"
-version = "0.13.4"
+version = "0.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0acc3549d5ce11c6901b3a676f2e2628684722197054d97cd0101ea174ed5cbd"
+checksum = "c4476d6a7c13c46ed9ff260bd0e1cf648dc37b13f483822e1ff2a431f0f6ee52"
 dependencies = [
  "cosmwasm-std",
  "cw-storage-plus",
@@ -1452,7 +1402,6 @@ dependencies = [
  "ff 0.11.0",
  "group 0.11.0",
  "lazy_static",
- "pemstore",
  "rand 0.8.5",
  "rand_chacha 0.3.1",
  "rand_core 0.6.3",
@@ -1504,9 +1453,9 @@ dependencies = [
 
 [[package]]
 name = "ed25519"
-version = "1.5.2"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e9c280362032ea4203659fc489832d0204ef09f247a0506f170dafcac08c369"
+checksum = "3d5c4b5e5959dc2c2b89918d8e2cc40fcdd623cef026ed09d2f0ee05199dc8e4"
 dependencies = [
  "serde",
  "signature",
@@ -1633,7 +1582,7 @@ dependencies = [
 
 [[package]]
 name = "explorer-api"
-version = "1.1.0"
+version = "1.0.1"
 dependencies = [
  "chrono",
  "clap 3.2.8",
@@ -2981,7 +2930,6 @@ dependencies = [
  "cosmwasm-std",
  "cw-utils",
  "cw3",
- "cw3-fixed-multisig",
  "cw4",
  "schemars",
  "serde",
@@ -3122,7 +3070,7 @@ dependencies = [
 
 [[package]]
 name = "nym-cli"
-version = "1.1.0"
+version = "1.0.0"
 dependencies = [
  "anyhow",
  "base64",
@@ -3139,7 +3087,6 @@ dependencies = [
  "pretty_env_logger",
  "serde",
  "serde_json",
- "tap",
  "tokio",
  "validator-client",
 ]
@@ -3165,7 +3112,6 @@ dependencies = [
  "rand 0.6.5",
  "serde",
  "serde_json",
- "tap",
  "thiserror",
  "time 0.3.14",
  "toml",
@@ -3176,10 +3122,9 @@ dependencies = [
 
 [[package]]
 name = "nym-client"
-version = "1.1.0"
+version = "1.0.2"
 dependencies = [
  "clap 3.2.8",
- "client-connections",
  "client-core",
  "coconut-interface",
  "completions",
@@ -3201,7 +3146,6 @@ dependencies = [
  "serde",
  "serde_json",
  "sled",
- "tap",
  "task",
  "thiserror",
  "tokio",
@@ -3216,7 +3160,7 @@ dependencies = [
 
 [[package]]
 name = "nym-gateway"
-version = "1.1.0"
+version = "1.0.2"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3263,7 +3207,7 @@ dependencies = [
 
 [[package]]
 name = "nym-mixnode"
-version = "1.1.0"
+version = "1.0.2"
 dependencies = [
  "anyhow",
  "bs58",
@@ -3305,11 +3249,10 @@ dependencies = [
 
 [[package]]
 name = "nym-network-requester"
-version = "1.1.0"
+version = "1.0.2"
 dependencies = [
  "async-trait",
  "clap 3.2.8",
- "client-connections",
  "completions",
  "dirs",
  "futures",
@@ -3353,10 +3296,9 @@ dependencies = [
 
 [[package]]
 name = "nym-socks5-client"
-version = "1.1.0"
+version = "1.0.2"
 dependencies = [
  "clap 3.2.8",
- "client-connections",
  "client-core",
  "coconut-interface",
  "completions",
@@ -3381,7 +3323,6 @@ dependencies = [
  "serde",
  "snafu 0.6.10",
  "socks5-requests",
- "tap",
  "task",
  "thiserror",
  "tokio",
@@ -3420,15 +3361,13 @@ dependencies = [
 
 [[package]]
 name = "nym-validator-api"
-version = "1.1.0"
+version = "1.0.2"
 dependencies = [
  "anyhow",
  "async-trait",
- "bs58",
  "cfg-if 1.0.0",
  "clap 3.2.8",
  "coconut-bandwidth-contract-common",
- "coconut-dkg-common",
  "coconut-interface",
  "config",
  "console-subscriber",
@@ -3440,7 +3379,6 @@ dependencies = [
  "cw-utils",
  "cw3",
  "dirs",
- "dkg",
  "dotenv",
  "futures",
  "gateway-client",
@@ -3454,7 +3392,6 @@ dependencies = [
  "nymcoconut",
  "nymsphinx",
  "okapi",
- "pemstore",
  "pin-project",
  "pretty_env_logger",
  "rand 0.7.3",
@@ -3507,19 +3444,16 @@ name = "nymcoconut"
 version = "0.5.0"
 dependencies = [
  "bincode",
- "bls12_381 0.6.0",
+ "bls12_381 0.5.0",
  "bs58",
  "criterion",
  "digest 0.9.0",
- "dkg",
  "doc-comment",
- "ff 0.11.0",
+ "ff 0.10.1",
  "getrandom 0.2.6",
- "group 0.11.0",
+ "group 0.10.0",
  "itertools",
- "pemstore",
  "rand 0.8.5",
- "rand_chacha 0.3.1",
  "serde",
  "serde_derive",
  "sha2 0.9.9",
@@ -4189,7 +4123,6 @@ name = "proxy-helpers"
 version = "0.1.0"
 dependencies = [
  "bytes",
- "client-connections",
  "futures",
  "log",
  "ordered-buffer",
@@ -5695,9 +5628,7 @@ checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369"
 name = "task"
 version = "0.1.0"
 dependencies = [
- "futures",
  "log",
- "thiserror",
  "tokio",
 ]
 
@@ -5837,18 +5768,18 @@ checksum = "b1141d4d61095b28419e22cb0bbf02755f5e54e0526f97f1e3d1d160e60885fb"
 
 [[package]]
 name = "thiserror"
-version = "1.0.37"
+version = "1.0.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10deb33631e3c9018b9baf9dcbbc4f737320d2b576bac10f6aefa048fa407e3e"
+checksum = "c53f98874615aea268107765aa1ed8f6116782501d18e53d08b471733bea6c85"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "1.0.37"
+version = "1.0.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "982d17546b47146b28f7c22e3d08465f6b8903d0ea13c1660d9d84a6e7adcdbb"
+checksum = "f8b463991b4eab2d801e724172285ec4195c650e8ec79b149e6c2a8e6dd3f783"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -6476,7 +6407,6 @@ dependencies = [
  "base64",
  "bip39",
  "coconut-bandwidth-contract-common",
- "coconut-dkg-common",
  "coconut-interface",
  "colored",
  "config",
@@ -6586,6 +6516,7 @@ version = "0.1.0"
 dependencies = [
  "contracts-common",
  "cosmwasm-std",
+ "log",
  "mixnet-contract-common",
  "schemars",
  "serde",

Cargo.toml

@@ -26,12 +26,10 @@ members = [
     "common/client-libs/gateway-client",
     "common/client-libs/mixnet-client",
     "common/client-libs/validator-client",
-    "common/client-connections",
     "common/coconut-interface",
     "common/commands",
     "common/config",
     "common/cosmwasm-smart-contracts/coconut-bandwidth-contract",
-    "common/cosmwasm-smart-contracts/coconut-dkg",
     "common/cosmwasm-smart-contracts/contracts-common",
     "common/cosmwasm-smart-contracts/mixnet-contract",
     "common/cosmwasm-smart-contracts/multisig-contract",

Makefile

@@ -92,9 +92,6 @@ build-wallet:
 build-connect:
 	cargo build --manifest-path nym-connect/Cargo.toml --workspace
 
-build-explorer-api:
-	cargo build --manifest-path explorer-api/Cargo.toml --workspace
-
 build-wasm-client:
 	cargo build --manifest-path clients/webassembly/Cargo.toml --workspace --target wasm32-unknown-unknown
 
@@ -125,3 +122,6 @@ mixnet-opt: wasm
 generate-typescript:
 	cd tools/ts-rs-cli && cargo run && cd ../..
 	yarn types:lint:fix
+
+run-validator-tests:
+	cd validator-api/tests/functional_test && yarn test

clients/client-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "client-core"
-version = "1.1.0"
+version = "1.0.1"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 edition = "2021"
 
@@ -14,13 +14,11 @@ log = "0.4"
 rand = { version = "0.7.3", features = ["wasm-bindgen"] }
 serde = { version = "1.0", features = ["derive"] }
 sled = { version = "0.34", optional = true }
-tap = "1.0.1"
 thiserror = "1.0.34"
 url = { version ="2.2", features = ["serde"] }
 
 # internal
 config = { path = "../../common/config" }
-client-connections = { path = "../../common/client-connections" }
 crypto = { path = "../../common/crypto" }
 gateway-client = { path = "../../common/client-libs/gateway-client" }
 #gateway-client = { path = "../../common/client-libs/gateway-client", default-features = false, features = ["wasm", "coconut"] }
@@ -30,6 +28,7 @@ nymsphinx = { path = "../../common/nymsphinx" }
 pemstore = { path = "../../common/pemstore" }
 topology = { path = "../../common/topology" }
 validator-client = { path = "../../common/client-libs/validator-client", default-features = false }
+tap = "1.0.1"
 
 tokio = { version = "1.21.2", features = ["time", "macros"]}
 
@@ -57,5 +56,4 @@ tempfile = "3.1.0"
 default = ["reply-surb"]
 wasm = ["gateway-client/wasm"]
 coconut = ["gateway-client/coconut", "gateway-requests/coconut"]
-reply-surb = ["sled"]
+reply-surb = ["sled"]
-

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

@@ -178,10 +178,6 @@ impl LoopCoverTrafficStream<OsRng> {
                     // This isn't a problem, if the channel is full means we're already sending the
                     // max amount of messages downstream can handle.
                     log::debug!("Failed to send cover message - channel full");
-                    // However it's still useful to alert the user that the gateway or the link to
-                    // the gateway can't keep up. Either due to insufficient bandwidth on the
-                    // client side, or that the gateway is overloaded.
-                    log::warn!("Failed to send: gateway appears to not keep up");
                 }
                 TrySendError::Closed(_) => {
                     log::warn!("Failed to send cover message - channel closed");
@@ -228,9 +224,7 @@ impl LoopCoverTrafficStream<OsRng> {
                     }
                 }
             }
-            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            assert!(shutdown.is_shutdown_poll());
-                .await
-                .expect("Task stopped without shutdown called");
             log::debug!("LoopCoverTrafficStream: Exiting");
         })
     }

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

@@ -1,9 +1,9 @@
-use client_connections::TransmissionLane;
+use futures::channel::mpsc;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::anonymous_replies::ReplySurb;
 
-pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
+pub type InputMessageSender = mpsc::UnboundedSender<InputMessage>;
-pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
+pub type InputMessageReceiver = mpsc::UnboundedReceiver<InputMessage>;
 
 #[derive(Debug)]
 pub enum InputMessage {
@@ -11,7 +11,6 @@ pub enum InputMessage {
         recipient: Recipient,
         data: Vec<u8>,
         with_reply_surb: bool,
-        lane: TransmissionLane,
     },
     Reply {
         reply_surb: ReplySurb,
@@ -20,17 +19,11 @@ pub enum InputMessage {
 }
 
 impl InputMessage {
-    pub fn new_fresh(
+    pub fn new_fresh(recipient: Recipient, data: Vec<u8>, with_reply_surb: bool) -> Self {
-        recipient: Recipient,
-        data: Vec<u8>,
-        with_reply_surb: bool,
-        lane: TransmissionLane,
-    ) -> Self {
         InputMessage::Fresh {
             recipient,
             data,
             with_reply_surb,
-            lane,
         }
     }
 

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

@@ -69,8 +69,6 @@ impl MixTrafficController {
 
     #[cfg(not(target_arch = "wasm32"))]
     pub fn start_with_shutdown(mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         spawn_future(async move {
             debug!("Started MixTrafficController with graceful shutdown support");
 
@@ -90,9 +88,7 @@ impl MixTrafficController {
                     }
                 }
             }
-            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            assert!(shutdown.is_shutdown_poll());
-                .await
-                .expect("Task stopped without shutdown called");
             log::debug!("MixTrafficController: Exiting");
         })
     }

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

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

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

@@ -33,7 +33,7 @@ impl AcknowledgementListener {
     }
 
     async fn on_ack(&mut self, ack_content: Vec<u8>) {
-        trace!("Received an ack");
+        debug!("Received an ack");
         let frag_id = match recover_identifier(&self.ack_key, &ack_content)
             .map(FragmentIdentifier::try_from_bytes)
         {
@@ -72,8 +72,6 @@ impl AcknowledgementListener {
 
     #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started AcknowledgementListener with graceful shutdown support");
 
         while !shutdown.is_shutdown() {
@@ -90,9 +88,7 @@ impl AcknowledgementListener {
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("AcknowledgementListener: Exiting");
     }
 

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

@@ -272,9 +272,7 @@ impl ActionController {
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("ActionController: Exiting");
     }
 

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

@@ -8,7 +8,7 @@ use crate::client::{
     real_messages_control::real_traffic_stream::{BatchRealMessageSender, RealMessage},
     topology_control::TopologyAccessor,
 };
-use client_connections::TransmissionLane;
+use futures::StreamExt;
 use log::*;
 use nymsphinx::anonymous_replies::ReplySurb;
 use nymsphinx::preparer::MessagePreparer;
@@ -104,7 +104,6 @@ where
         content: Vec<u8>,
         with_reply_surb: bool,
     ) -> Option<Vec<RealMessage>> {
-        log::trace!("handling msg size: {}", content.len());
         let topology_permit = self.topology_access.get_read_permit().await;
         let topology = match topology_permit
             .try_get_valid_topology_ref(&self.ack_recipient, Some(&recipient))
@@ -165,44 +164,37 @@ where
     }
 
     async fn on_input_message(&mut self, msg: InputMessage) {
-        let (real_messages, lane) = match msg {
+        let real_messages = match msg {
             InputMessage::Fresh {
                 recipient,
                 data,
                 with_reply_surb,
-                lane,
+            } => {
-            } => (
                 self.handle_fresh_message(recipient, data, with_reply_surb)
-                    .await,
-                lane,
-            ),
-            InputMessage::Reply { reply_surb, data } => (
-                self.handle_reply(reply_surb, data)
                     .await
-                    .map(|message| vec![message]),
+            }
-                TransmissionLane::Reply,
+            InputMessage::Reply { reply_surb, data } => self
-            ),
+                .handle_reply(reply_surb, data)
+                .await
+                .map(|message| vec![message]),
         };
 
         // there's no point in trying to send nothing
         if let Some(real_messages) = real_messages {
             // tells real message sender (with the poisson timer) to send this to the mix network
             self.real_message_sender
-                .send((real_messages, lane))
+                .unbounded_send(real_messages)
-                .await
+                .unwrap();
-                .expect("BatchRealMessageReceiver has stopped receiving!");
         }
     }
 
     #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started InputMessageListener with graceful shutdown support");
 
         while !shutdown.is_shutdown() {
             tokio::select! {
-                input_msg = self.input_receiver.recv() => match input_msg {
+                input_msg = self.input_receiver.next() => match input_msg {
                     Some(input_msg) => {
                         self.on_input_message(input_msg).await;
                     },
@@ -216,16 +208,14 @@ where
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("InputMessageListener: Exiting");
     }
 
     #[cfg(target_arch = "wasm32")]
     pub(super) async fn run(&mut self) {
         debug!("Started InputMessageListener without graceful shutdown support");
-        while let Some(input_msg) = self.input_receiver.recv().await {
+        while let Some(input_msg) = self.input_receiver.next().await {
             self.on_input_message(input_msg).await;
         }
     }

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

@@ -1,21 +1,17 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use super::{
+use super::action_controller::{Action, ActionSender};
-    action_controller::{Action, ActionSender},
+use super::PendingAcknowledgement;
-    PendingAcknowledgement, RetransmissionRequestReceiver,
+use super::RetransmissionRequestReceiver;
-};
 use crate::client::{
     real_messages_control::real_traffic_stream::{BatchRealMessageSender, RealMessage},
     topology_control::TopologyAccessor,
 };
-
-use client_connections::TransmissionLane;
 use futures::StreamExt;
 use log::*;
-use nymsphinx::{
+use nymsphinx::preparer::MessagePreparer;
-    acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer,
+use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
-};
 use rand::{CryptoRng, Rng};
 use std::sync::{Arc, Weak};
 
@@ -117,18 +113,15 @@ where
 
         // send to `OutQueueControl` to eventually send to the mix network
         self.real_message_sender
-            .send((
+            .unbounded_send(vec![RealMessage::new(
-                vec![RealMessage::new(prepared_fragment.mix_packet, frag_id)],
+                prepared_fragment.mix_packet,
-                TransmissionLane::Retransmission,
+                frag_id,
-            ))
+            )])
-            .await
+            .unwrap();
-            .expect("BatchRealMessageReceiver has stopped receiving!");
     }
 
     #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started RetransmissionRequestListener with graceful shutdown support");
 
         while !shutdown.is_shutdown() {
@@ -145,9 +138,7 @@ where
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("RetransmissionRequestListener: Exiting");
     }
 

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

@@ -8,15 +8,12 @@
 use self::{
     acknowledgement_control::AcknowledgementController, real_traffic_stream::OutQueueControl,
 };
-use crate::{
+use crate::client::real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors;
-    client::{
+use crate::client::{
-        inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
+    inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
-        real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
+    topology_control::TopologyAccessor,
-        topology_control::TopologyAccessor,
-    },
-    spawn_future,
 };
-use client_connections::{ConnectionCommandReceiver, LaneQueueLengths};
+use crate::spawn_future;
 use futures::channel::mpsc;
 use gateway_client::AcknowledgementReceiver;
 use log::*;
@@ -106,7 +103,6 @@ where
 // obviously when we finally make shared rng that is on 'higher' level, this should become
 // generic `R`
 impl RealMessagesController<OsRng> {
-    #[allow(clippy::too_many_arguments)]
     pub fn new(
         config: Config,
         ack_receiver: AcknowledgementReceiver,
@@ -114,12 +110,10 @@ impl RealMessagesController<OsRng> {
         mix_sender: BatchMixMessageSender,
         topology_access: TopologyAccessor,
         #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
-        lane_queue_lengths: LaneQueueLengths,
-        client_connection_rx: ConnectionCommandReceiver,
     ) -> Self {
         let rng = OsRng;
 
-        let (real_message_sender, real_message_receiver) = tokio::sync::mpsc::channel(1);
+        let (real_message_sender, real_message_receiver) = mpsc::unbounded();
         let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded();
 
         let ack_controller_connectors = AcknowledgementControllerConnectors::new(
@@ -165,8 +159,6 @@ impl RealMessagesController<OsRng> {
             rng,
             config.self_recipient,
             topology_access,
-            lane_queue_lengths,
-            client_connection_rx,
         );
 
         RealMessagesController {

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

@@ -4,9 +4,7 @@
 use crate::client::mix_traffic::BatchMixMessageSender;
 use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender;
 use crate::client::topology_control::TopologyAccessor;
-use client_connections::{
+use futures::channel::mpsc;
-    ConnectionCommand, ConnectionCommandReceiver, ConnectionId, LaneQueueLengths, TransmissionLane,
-};
 use futures::task::{Context, Poll};
 use futures::{Future, Stream, StreamExt};
 use log::*;
@@ -18,6 +16,7 @@ use nymsphinx::forwarding::packet::MixPacket;
 use nymsphinx::params::PacketSize;
 use nymsphinx::utils::sample_poisson_duration;
 use rand::{CryptoRng, Rng};
+use std::collections::VecDeque;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
@@ -28,22 +27,22 @@ use tokio::time;
 #[cfg(target_arch = "wasm32")]
 use wasm_timer;
 
-use self::{
+// The minimum time between increasing the average delay between packets. If we hit the ceiling in
-    sending_delay_controller::SendingDelayController, transmission_buffer::TransmissionBuffer,
+// the available buffer space we want to take somewhat swift action, but we still need to give a
-};
+// short time to give the channel a chance reduce pressure.
-
+const INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 1;
-mod sending_delay_controller;
+// The minimum time between decreasing the average delay between packets. We don't want to change
-mod transmission_buffer;
+// to quickly to keep things somewhat stable. Also there are buffers downstreams meaning we need to
-
+// wait a little to see the effect before we decrease further.
-#[cfg(not(target_arch = "wasm32"))]
+const DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 30;
-fn get_time_now() -> time::Instant {
+// If we enough time passes without any sign of backpressure in the channel, we can consider
-    time::Instant::now()
+// lowering the average delay. The goal is to keep somewhat stable, rather than maxing out
-}
+// bandwidth at all times.
-
+const ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS: u64 = 30;
-#[cfg(target_arch = "wasm32")]
+// The maximum multiplier we apply to the base average Poisson delay.
-fn get_time_now() -> wasm_timer::Instant {
+const MAX_DELAY_MULTIPLIER: u32 = 6;
-    wasm_timer::Instant::now()
+// The minium multiplier we apply to the base average Poisson delay.
-}
+const MIN_DELAY_MULTIPLIER: u32 = 1;
 
 /// Configurable parameters of the `OutQueueControl`
 pub(crate) struct Config {
@@ -86,6 +85,101 @@ impl Config {
     }
 }
 
+struct SendingDelayController {
+    /// Multiply the average sending delay.
+    /// This is normally set to unity, but if we detect backpressure we increase this
+    /// multiplier. We use discrete steps.
+    current_multiplier: u32,
+
+    /// Maximum delay multiplier
+    upper_bound: u32,
+
+    /// Minimum delay multiplier
+    lower_bound: u32,
+
+    /// To make sure we don't change the multiplier to fast, we limit a change to some duration
+    #[cfg(not(target_arch = "wasm32"))]
+    time_when_changed: time::Instant,
+
+    #[cfg(target_arch = "wasm32")]
+    time_when_changed: wasm_timer::Instant,
+
+    /// If we have a long enough time without any backpressure detected we try reducing the sending
+    /// delay multiplier
+    #[cfg(not(target_arch = "wasm32"))]
+    time_when_backpressure_detected: time::Instant,
+
+    #[cfg(target_arch = "wasm32")]
+    time_when_backpressure_detected: wasm_timer::Instant,
+}
+
+#[cfg(not(target_arch = "wasm32"))]
+fn get_time_now() -> time::Instant {
+    time::Instant::now()
+}
+
+#[cfg(target_arch = "wasm32")]
+fn get_time_now() -> wasm_timer::Instant {
+    wasm_timer::Instant::now()
+}
+
+impl SendingDelayController {
+    fn new(lower_bound: u32, upper_bound: u32) -> Self {
+        assert!(lower_bound <= upper_bound);
+        let now = get_time_now();
+        SendingDelayController {
+            current_multiplier: MIN_DELAY_MULTIPLIER,
+            upper_bound,
+            lower_bound,
+            time_when_changed: now,
+            time_when_backpressure_detected: now,
+        }
+    }
+
+    fn current_multiplier(&self) -> u32 {
+        self.current_multiplier
+    }
+
+    fn increase_delay_multiplier(&mut self) {
+        self.current_multiplier =
+            (self.current_multiplier + 1).clamp(self.lower_bound, self.upper_bound);
+        self.time_when_changed = get_time_now();
+        log::debug!(
+            "Increasing sending delay multiplier to: {}",
+            self.current_multiplier
+        );
+    }
+
+    fn decrease_delay_multiplier(&mut self) {
+        self.current_multiplier =
+            (self.current_multiplier - 1).clamp(self.lower_bound, self.upper_bound);
+        self.time_when_changed = get_time_now();
+        log::debug!(
+            "Decreasing sending delay multiplier to: {}",
+            self.current_multiplier
+        );
+    }
+
+    fn record_backpressure_detected(&mut self) {
+        self.time_when_backpressure_detected = get_time_now();
+    }
+
+    fn not_increased_delay_recently(&self) -> bool {
+        get_time_now()
+            > self.time_when_changed + Duration::from_secs(INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS)
+    }
+
+    fn is_sending_reliable(&self) -> bool {
+        let now = get_time_now();
+        let delay_change_interval = Duration::from_secs(DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS);
+        let acceptable_time_without_backpressure =
+            Duration::from_secs(ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS);
+
+        now > self.time_when_backpressure_detected + acceptable_time_without_backpressure
+            && now > self.time_when_changed + delay_change_interval
+    }
+}
+
 pub(crate) struct OutQueueControl<R>
 where
     R: CryptoRng + Rng,
@@ -109,7 +203,7 @@ where
 
     // To make sure we don't overload the mix_tx channel, we limit the rate we are pushing
     // messages.
-    sending_delay_controller: SendingDelayController,
+    sending_rate_controller: SendingDelayController,
 
     /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them
     /// out to the network without any further delays.
@@ -128,19 +222,10 @@ where
     /// Accessor to the common instance of network topology.
     topology_access: TopologyAccessor,
 
-    /// Buffer containing all incoming real messages keyed by transmission lane, that we will send
+    /// Buffer containing all real messages received. It is first exhausted before more are pulled.
-    /// out to the mixnet.
+    received_buffer: VecDeque<RealMessage>,
-    transmission_buffer: TransmissionBuffer,
-
-    /// Incoming channel for being notified of closed connections, so that we can close lanes
-    /// corresponding to connections. To avoid sending traffic unnecessary
-    client_connection_rx: ConnectionCommandReceiver,
-
-    /// Report queue lengths so that upstream can backoff sending data, and keep connections open.
-    lane_queue_lengths: LaneQueueLengths,
 }
 
-#[derive(Debug)]
 pub(crate) struct RealMessage {
     mix_packet: MixPacket,
     fragment_id: FragmentIdentifier,
@@ -157,9 +242,8 @@ impl RealMessage {
 
 // messages are already prepared, etc. the real point of it is to forward it to mix_traffic
 // after sufficient delay
-pub(crate) type BatchRealMessageSender =
+pub(crate) type BatchRealMessageSender = mpsc::UnboundedSender<Vec<RealMessage>>;
-    tokio::sync::mpsc::Sender<(Vec<RealMessage>, TransmissionLane)>;
+type BatchRealMessageReceiver = mpsc::UnboundedReceiver<Vec<RealMessage>>;
-type BatchRealMessageReceiver = tokio::sync::mpsc::Receiver<(Vec<RealMessage>, TransmissionLane)>;
 
 pub(crate) enum StreamMessage {
     Cover,
@@ -182,23 +266,22 @@ where
         rng: R,
         our_full_destination: Recipient,
         topology_access: TopologyAccessor,
-        lane_queue_lengths: LaneQueueLengths,
-        client_connection_rx: ConnectionCommandReceiver,
     ) -> Self {
         OutQueueControl {
             config,
             ack_key,
             sent_notifier,
             next_delay: None,
-            sending_delay_controller: Default::default(),
+            sending_rate_controller: SendingDelayController::new(
+                MIN_DELAY_MULTIPLIER,
+                MAX_DELAY_MULTIPLIER,
+            ),
             mix_tx,
             real_receiver,
             our_full_destination,
             rng,
             topology_access,
-            transmission_buffer: Default::default(),
+            received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly
-            client_connection_rx,
-            lane_queue_lengths,
         }
     }
 
@@ -254,7 +337,7 @@ where
         };
 
         if let Err(err) = self.mix_tx.send(vec![next_message]).await {
-            log::error!("Failed to send: {}", err);
+            log::error!("Failed to send - channel closed: {}", err);
         }
 
         // notify ack controller about sending our message only after we actually managed to push it
@@ -263,10 +346,6 @@ where
             self.sent_notify(fragment_id);
         }
 
-        // In addition to closing connections on receiving messages throught client_connection_rx,
-        // also close connections when sufficiently stale.
-        self.transmission_buffer.prune_stale_connections();
-
         // JS: Not entirely sure why or how it fixes stuff, but without the yield call,
         // the UnboundedReceiver [of mix_rx] will not get a chance to read anything
         // JS2: Basically it was the case that with high enough rate, the stream had already a next value
@@ -278,71 +357,44 @@ where
         tokio::task::yield_now().await;
     }
 
-    fn on_close_connection(&mut self, connection_id: ConnectionId) {
-        log::debug!("Removing lane for connection: {connection_id}");
-        self.transmission_buffer
-            .remove(&TransmissionLane::ConnectionId(connection_id));
-    }
-
     fn current_average_message_sending_delay(&self) -> Duration {
         self.config.average_message_sending_delay
-            * self.sending_delay_controller.current_multiplier()
+            * self.sending_rate_controller.current_multiplier()
     }
 
     fn adjust_current_average_message_sending_delay(&mut self) {
         let used_slots = self.mix_tx.max_capacity() - self.mix_tx.capacity();
         log::trace!(
             "used_slots: {used_slots}, current_multiplier: {}",
-            self.sending_delay_controller.current_multiplier()
+            self.sending_rate_controller.current_multiplier()
         );
 
         // Even just a single used slot is enough to signal backpressure
         if used_slots > 0 {
             log::trace!("Backpressure detected");
-            self.sending_delay_controller.record_backpressure_detected();
+            self.sending_rate_controller.record_backpressure_detected();
         }
 
         // If the buffer is running out, slow down the sending rate
         if self.mix_tx.capacity() == 0
-            && self.sending_delay_controller.not_increased_delay_recently()
+            && self.sending_rate_controller.not_increased_delay_recently()
         {
-            self.sending_delay_controller.increase_delay_multiplier();
+            self.sending_rate_controller.increase_delay_multiplier();
         }
 
         // Very carefully step up the sending rate in case it seems like we can solidly handle the
         // current rate.
-        if self.sending_delay_controller.is_sending_reliable() {
+        if self.sending_rate_controller.is_sending_reliable() {
-            self.sending_delay_controller.decrease_delay_multiplier();
+            self.sending_rate_controller.decrease_delay_multiplier();
         }
     }
 
-    fn pop_next_message(&mut self) -> Option<RealMessage> {
-        // Pop the next message from the transmission buffer
-        let (lane, real_next) = self.transmission_buffer.pop_next_message_at_random()?;
-
-        // Update the published queue length
-        let lane_length = self.transmission_buffer.lane_length(&lane);
-        self.lane_queue_lengths.set(&lane, lane_length);
-
-        Some(real_next)
-    }
-
     fn poll_poisson(&mut self, cx: &mut Context<'_>) -> Poll<Option<StreamMessage>> {
         // The average delay could change depending on if backpressure in the downstream channel
         // (mix_tx) was detected.
         self.adjust_current_average_message_sending_delay();
         let avg_delay = self.current_average_message_sending_delay();
 
-        // Start by checking if we have any incoming messages about closed connections
-        // NOTE: this feels a bit iffy, the `OutQueueControl` is getting ripe for a rewrite to
-        // something simpler.
-        if let Poll::Ready(Some(id)) = Pin::new(&mut self.client_connection_rx).poll_next(cx) {
-            match id {
-                ConnectionCommand::Close(id) => self.on_close_connection(id),
-                ConnectionCommand::ActiveConnections(_) => panic!(),
-            }
-        }
-
         if let Some(ref mut next_delay) = &mut self.next_delay {
             // it is not yet time to return a message
             if next_delay.as_mut().poll(cx).is_pending() {
@@ -367,32 +419,28 @@ where
                 next_delay.as_mut().reset(next_poisson_delay);
             }
 
-            // On every iteration we get new messages from upstream. Given that these come bunched
+            // check if we have anything immediately available
-            // in `Vec`, this ensures that on average we will fetch messages faster than we can
+            if let Some(real_available) = self.received_buffer.pop_front() {
-            // send, which is a condition for being able to multiplex sphinx packets from multiple
+                return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available))));
-            // data streams.
+            }
-            match Pin::new(&mut self.real_receiver).poll_recv(cx) {
+
+            // decide what kind of message to send
+            match Pin::new(&mut self.real_receiver).poll_next(cx) {
                 // in the case our real message channel stream was closed, we should also indicate we are closed
                 // (and whoever is using the stream should panic)
                 Poll::Ready(None) => Poll::Ready(None),
 
-                Poll::Ready(Some((real_messages, conn_id))) => {
+                // if there are more messages available, return first one and store the rest
-                    log::trace!("handling real_messages: size: {}", real_messages.len());
+                Poll::Ready(Some(real_messages)) => {
-
+                    self.received_buffer = real_messages.into();
-                    self.transmission_buffer.store(&conn_id, real_messages);
+                    // we MUST HAVE received at least ONE message
-                    let real_next = self.pop_next_message().expect("Just stored one");
+                    Poll::Ready(Some(StreamMessage::Real(Box::new(
-
+                        self.received_buffer.pop_front().unwrap(),
-                    Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
+                    ))))
                 }
 
-                Poll::Pending => {
+                // otherwise construct a dummy one
-                    if let Some(real_next) = self.pop_next_message() {
+                Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)),
-                        Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
-                    } else {
-                        // otherwise construct a dummy one
-                        Poll::Ready(Some(StreamMessage::Cover))
-                    }
-                }
             }
         } else {
             // we never set an initial delay - let's do it now
@@ -414,36 +462,32 @@ where
     }
 
     fn poll_immediate(&mut self, cx: &mut Context<'_>) -> Poll<Option<StreamMessage>> {
-        // Start by checking if we have any incoming messages about closed connections
+        // check if we have anything immediately available
-        if let Poll::Ready(Some(id)) = Pin::new(&mut self.client_connection_rx).poll_next(cx) {
+        if let Some(real_available) = self.received_buffer.pop_front() {
-            match id {
+            // if there are more messages immediately available, notify the runtime
-                ConnectionCommand::Close(id) => self.on_close_connection(id),
+            // because we should be polled again
-                ConnectionCommand::ActiveConnections(_) => panic!(),
+            if !self.received_buffer.is_empty() {
+                cx.waker().wake_by_ref()
             }
+            return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available))));
         }
 
-        match Pin::new(&mut self.real_receiver).poll_recv(cx) {
+        match Pin::new(&mut self.real_receiver).poll_next(cx) {
             // in the case our real message channel stream was closed, we should also indicate we are closed
             // (and whoever is using the stream should panic)
             Poll::Ready(None) => Poll::Ready(None),
 
-            Poll::Ready(Some((real_messages, conn_id))) => {
+            // if there are more messages available, return first one and store the rest
-                log::trace!("handling real_messages: size: {}", real_messages.len());
+            Poll::Ready(Some(real_messages)) => {
-
+                self.received_buffer = real_messages.into();
-                // First store what we got for the given connection id
+                // we MUST HAVE received at least ONE message
-                self.transmission_buffer.store(&conn_id, real_messages);
+                Poll::Ready(Some(StreamMessage::Real(Box::new(
-                let real_next = self.pop_next_message().expect("we just added one");
+                    self.received_buffer.pop_front().unwrap(),
-
+                ))))
-                Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
             }
 
-            Poll::Pending => {
+            // if there's nothing, then there's nothing
-                if let Some(real_next) = self.pop_next_message() {
+            Poll::Pending => Poll::Pending,
-                    Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
-                } else {
-                    Poll::Pending
-                }
-            }
         }
     }
 
@@ -458,73 +502,28 @@ where
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
-    fn log_status(&self) {
-        let packets = self.transmission_buffer.total_size();
-        let backlog = self.transmission_buffer.total_size_in_bytes() as f64 / 1024.0;
-        let lanes = self.transmission_buffer.num_lanes();
-        let mult = self.sending_delay_controller.current_multiplier();
-        let delay = self.current_average_message_sending_delay().as_millis();
-        let status_str = if self.config.disable_poisson_packet_distribution {
-            format!(
-                "Status: {lanes} lanes, backlog: {:.2} kiB ({packets}), no delay",
-                backlog
-            )
-        } else {
-            format!(
-                "Status: {lanes} lanes, backlog: {:.2} kiB ({packets}), avg delay: {}ms ({mult})",
-                backlog, delay
-            )
-        };
-        if packets > 1000 {
-            log::warn!("{status_str}");
-        } else if packets > 0 {
-            log::info!("{status_str}");
-        } else {
-            log::debug!("{status_str}");
-        }
-    }
-
-    #[cfg(not(target_arch = "wasm32"))]
-    fn log_status_infrequent(&self) {
-        if self.sending_delay_controller.current_multiplier() > 1 {
-            log::warn!(
-                "Unable to send packets fast enough - sending delay multiplier set to: {}",
-                self.sending_delay_controller.current_multiplier()
-            );
-        }
-    }
-
     #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
         debug!("Started OutQueueControl with graceful shutdown support");
 
-        let mut status_timer = tokio::time::interval(Duration::from_secs(5));
-        let mut infrequent_status_timer = tokio::time::interval(Duration::from_secs(60));
-
         while !shutdown.is_shutdown() {
             tokio::select! {
                 biased;
                 _ = shutdown.recv() => {
                     log::trace!("OutQueueControl: Received shutdown");
                 }
-                _ = status_timer.tick() => {
+                next_message = self.next() => match next_message {
-                    self.log_status();
+                    Some(next_message) => {
-                }
+                        self.on_message(next_message).await;
-                _ = infrequent_status_timer.tick() => {
+                    },
-                    self.log_status_infrequent();
+                    None => {
-                }
+                        log::trace!("OutQueueControl: Stopping since channel closed");
-                next_message = self.next() => if let Some(next_message) = next_message {
+                        break;
-                    self.on_message(next_message).await;
+                    }
-                } else {
-                    log::trace!("OutQueueControl: Stopping since channel closed");
-                    break;
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("OutQueueControl: Exiting");
     }
 

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

@@ -1,124 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use super::get_time_now;
-use std::time::Duration;
-
-#[cfg(not(target_arch = "wasm32"))]
-use tokio::time;
-#[cfg(target_arch = "wasm32")]
-use wasm_timer;
-
-// The minimum time between increasing the average delay between packets. If we hit the ceiling in
-// the available buffer space we want to take somewhat swift action, but we still need to give a
-// short time to give the channel a chance reduce pressure.
-const INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 1;
-// The minimum time between decreasing the average delay between packets. We don't want to change
-// to quickly to keep things somewhat stable. Also there are buffers downstreams meaning we need to
-// wait a little to see the effect before we decrease further.
-const DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS: u64 = 30;
-// If we enough time passes without any sign of backpressure in the channel, we can consider
-// lowering the average delay. The goal is to keep somewhat stable, rather than maxing out
-// bandwidth at all times.
-const ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS: u64 = 30;
-// The maximum multiplier we apply to the base average Poisson delay.
-const MAX_DELAY_MULTIPLIER: u32 = 6;
-// The minium multiplier we apply to the base average Poisson delay.
-const MIN_DELAY_MULTIPLIER: u32 = 1;
-
-pub(crate) struct SendingDelayController {
-    /// Multiply the average sending delay.
-    /// This is normally set to unity, but if we detect backpressure we increase this
-    /// multiplier. We use discrete steps.
-    current_multiplier: u32,
-
-    /// Maximum delay multiplier
-    upper_bound: u32,
-
-    /// Minimum delay multiplier
-    lower_bound: u32,
-
-    /// To make sure we don't change the multiplier to fast, we limit a change to some duration
-    #[cfg(not(target_arch = "wasm32"))]
-    time_when_changed: time::Instant,
-
-    #[cfg(target_arch = "wasm32")]
-    time_when_changed: wasm_timer::Instant,
-
-    /// If we have a long enough time without any backpressure detected we try reducing the sending
-    /// delay multiplier
-    #[cfg(not(target_arch = "wasm32"))]
-    time_when_backpressure_detected: time::Instant,
-
-    #[cfg(target_arch = "wasm32")]
-    time_when_backpressure_detected: wasm_timer::Instant,
-}
-
-impl Default for SendingDelayController {
-    fn default() -> Self {
-        SendingDelayController::new(MIN_DELAY_MULTIPLIER, MAX_DELAY_MULTIPLIER)
-    }
-}
-
-impl SendingDelayController {
-    pub(crate) fn new(lower_bound: u32, upper_bound: u32) -> Self {
-        assert!(lower_bound <= upper_bound);
-        let now = get_time_now();
-        SendingDelayController {
-            current_multiplier: MIN_DELAY_MULTIPLIER,
-            upper_bound,
-            lower_bound,
-            time_when_changed: now,
-            time_when_backpressure_detected: now,
-        }
-    }
-
-    pub(crate) fn current_multiplier(&self) -> u32 {
-        self.current_multiplier
-    }
-
-    pub(crate) fn increase_delay_multiplier(&mut self) {
-        if self.current_multiplier < self.upper_bound {
-            self.current_multiplier =
-                (self.current_multiplier + 1).clamp(self.lower_bound, self.upper_bound);
-            self.time_when_changed = get_time_now();
-            log::warn!(
-                "Increasing sending delay multiplier to: {}",
-                self.current_multiplier
-            );
-        } else {
-            log::warn!("Trying to increase delay multipler higher than allowed");
-        }
-    }
-
-    pub(crate) fn decrease_delay_multiplier(&mut self) {
-        if self.current_multiplier > self.lower_bound {
-            self.current_multiplier =
-                (self.current_multiplier - 1).clamp(self.lower_bound, self.upper_bound);
-            self.time_when_changed = get_time_now();
-            log::debug!(
-                "Decreasing sending delay multiplier to: {}",
-                self.current_multiplier
-            );
-        }
-    }
-
-    pub(crate) fn record_backpressure_detected(&mut self) {
-        self.time_when_backpressure_detected = get_time_now();
-    }
-
-    pub(crate) fn not_increased_delay_recently(&self) -> bool {
-        get_time_now()
-            > self.time_when_changed + Duration::from_secs(INCREASE_DELAY_MIN_CHANGE_INTERVAL_SECS)
-    }
-
-    pub(crate) fn is_sending_reliable(&self) -> bool {
-        let now = get_time_now();
-        let delay_change_interval = Duration::from_secs(DECREASE_DELAY_MIN_CHANGE_INTERVAL_SECS);
-        let acceptable_time_without_backpressure =
-            Duration::from_secs(ACCEPTABLE_TIME_WITHOUT_BACKPRESSURE_SECS);
-
-        now > self.time_when_backpressure_detected + acceptable_time_without_backpressure
-            && now > self.time_when_changed + delay_change_interval
-    }
-}

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

@@ -1,211 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use client_connections::TransmissionLane;
-use rand::seq::SliceRandom;
-use std::{
-    collections::{HashMap, HashSet, VecDeque},
-    time::Duration,
-};
-
-#[cfg(not(target_arch = "wasm32"))]
-use tokio::time;
-
-#[cfg(target_arch = "wasm32")]
-use wasm_timer;
-
-use super::{get_time_now, RealMessage};
-
-// The number of lanes included in the oldest set. Used when we need to prioritize traffic.
-const OLDEST_LANE_SET_SIZE: usize = 5;
-// As a way of prune connections we also check for timeouts.
-const MSG_CONSIDERED_STALE_AFTER_SECS: u64 = 10 * 60;
-
-#[derive(Default)]
-pub(crate) struct TransmissionBuffer {
-    buffer: HashMap<TransmissionLane, LaneBufferEntry>,
-}
-
-impl TransmissionBuffer {
-    #[allow(unused)]
-    pub(crate) fn is_empty(&self) -> bool {
-        self.buffer.is_empty()
-    }
-
-    pub(crate) fn remove(&mut self, lane: &TransmissionLane) -> Option<LaneBufferEntry> {
-        self.buffer.remove(lane)
-    }
-
-    #[cfg(not(target_arch = "wasm32"))]
-    pub(crate) fn num_lanes(&self) -> usize {
-        self.buffer.keys().count()
-    }
-
-    pub(crate) fn lane_length(&self, lane: &TransmissionLane) -> Option<usize> {
-        self.buffer.get(lane).map(LaneBufferEntry::len)
-    }
-
-    #[allow(unused)]
-    pub(crate) fn connections(&self) -> HashSet<u64> {
-        self.buffer
-            .keys()
-            .filter_map(|lane| match lane {
-                TransmissionLane::ConnectionId(id) => Some(id),
-                _ => None,
-            })
-            .copied()
-            .collect()
-    }
-
-    #[cfg(not(target_arch = "wasm32"))]
-    pub(crate) fn total_size(&self) -> usize {
-        self.buffer.values().map(LaneBufferEntry::len).sum()
-    }
-
-    #[cfg(not(target_arch = "wasm32"))]
-    pub(crate) fn total_size_in_bytes(&self) -> usize {
-        self.buffer
-            .values()
-            .map(|lane_buffer_entry| {
-                lane_buffer_entry
-                    .real_messages
-                    .iter()
-                    .map(|real_message| real_message.mix_packet.sphinx_packet().len())
-                    .sum::<usize>()
-            })
-            .sum()
-    }
-
-    fn get_oldest_set(&self) -> Vec<TransmissionLane> {
-        let mut buffer: Vec<_> = self
-            .buffer
-            .iter()
-            .map(|(k, v)| (k, v.messages_transmitted))
-            .collect();
-        buffer.sort_by_key(|v| v.1);
-        buffer
-            .iter()
-            .rev()
-            .map(|(k, _)| *k)
-            .take(OLDEST_LANE_SET_SIZE)
-            .copied()
-            .collect()
-    }
-
-    pub(crate) fn store(&mut self, lane: &TransmissionLane, real_messages: Vec<RealMessage>) {
-        if let Some(lane_buffer_entry) = self.buffer.get_mut(lane) {
-            lane_buffer_entry.append(real_messages);
-        } else {
-            self.buffer
-                .insert(*lane, LaneBufferEntry::new(real_messages));
-        }
-    }
-
-    fn pick_random_lane(&self) -> Option<&TransmissionLane> {
-        let lanes: Vec<&TransmissionLane> = self.buffer.keys().collect();
-        lanes.choose(&mut rand::thread_rng()).copied()
-    }
-
-    fn pick_random_small_lane(&self) -> Option<&TransmissionLane> {
-        let lanes: Vec<&TransmissionLane> = self
-            .buffer
-            .iter()
-            .filter(|(_, v)| v.is_small())
-            .map(|(k, _)| k)
-            .collect();
-        lanes.choose(&mut rand::thread_rng()).copied()
-    }
-
-    fn pick_random_old_lane(&self) -> Option<TransmissionLane> {
-        let lanes = self.get_oldest_set();
-        lanes.choose(&mut rand::thread_rng()).copied()
-    }
-
-    fn pop_front_from_lane(&mut self, lane: &TransmissionLane) -> Option<RealMessage> {
-        let real_msgs_queued = self.buffer.get_mut(lane)?;
-        let real_next = real_msgs_queued.pop_front()?;
-        real_msgs_queued.messages_transmitted += 1;
-        if real_msgs_queued.is_empty() {
-            self.buffer.remove(lane);
-        }
-        Some(real_next)
-    }
-
-    pub(crate) fn pop_next_message_at_random(&mut self) -> Option<(TransmissionLane, RealMessage)> {
-        if self.buffer.is_empty() {
-            return None;
-        }
-
-        // Very basic heuristic where we prioritize according to small lanes first, the older lanes
-        // to try to finish lanes when possible, then the rest.
-        let lane = if let Some(small_lane) = self.pick_random_small_lane() {
-            *small_lane
-        } else if let Some(old_lane) = self.pick_random_old_lane() {
-            old_lane
-        } else {
-            *self.pick_random_lane()?
-        };
-
-        let msg = self.pop_front_from_lane(&lane)?;
-        log::trace!("picking to send from lane: {:?}", lane);
-        Some((lane, msg))
-    }
-
-    pub(crate) fn prune_stale_connections(&mut self) {
-        let stale_entries: Vec<_> = self
-            .buffer
-            .iter()
-            .filter_map(|(lane, entry)| if entry.is_stale() { Some(lane) } else { None })
-            .copied()
-            .collect();
-
-        for lane in stale_entries {
-            self.remove(&lane);
-        }
-    }
-}
-
-pub(crate) struct LaneBufferEntry {
-    pub real_messages: VecDeque<RealMessage>,
-    pub messages_transmitted: usize,
-    #[cfg(not(target_arch = "wasm32"))]
-    pub time_for_last_activity: time::Instant,
-    #[cfg(target_arch = "wasm32")]
-    pub time_for_last_activity: wasm_timer::Instant,
-}
-
-impl LaneBufferEntry {
-    fn new(real_messages: Vec<RealMessage>) -> Self {
-        LaneBufferEntry {
-            real_messages: real_messages.into(),
-            messages_transmitted: 0,
-            time_for_last_activity: get_time_now(),
-        }
-    }
-
-    fn append(&mut self, real_messages: Vec<RealMessage>) {
-        self.real_messages.append(&mut real_messages.into());
-        self.time_for_last_activity = get_time_now();
-    }
-
-    fn pop_front(&mut self) -> Option<RealMessage> {
-        self.real_messages.pop_front()
-    }
-
-    fn is_small(&self) -> bool {
-        self.real_messages.len() < 100
-    }
-
-    fn is_stale(&self) -> bool {
-        get_time_now() - self.time_for_last_activity
-            > Duration::from_secs(MSG_CONSIDERED_STALE_AFTER_SECS)
-    }
-
-    fn len(&self) -> usize {
-        self.real_messages.len()
-    }
-
-    fn is_empty(&self) -> bool {
-        self.real_messages.is_empty()
-    }
-}

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

@@ -208,7 +208,7 @@ impl ReceivedMessagesBuffer {
     }
 
     async fn handle_new_received(&mut self, msgs: Vec<Vec<u8>>) {
-        trace!(
+        debug!(
             "Processing {:?} new message that might get added to the buffer!",
             msgs.len()
         );
@@ -322,8 +322,6 @@ impl RequestReceiver {
 
     #[cfg(not(target_arch = "wasm32"))]
     async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started RequestReceiver with graceful shutdown support");
         while !shutdown.is_shutdown() {
             tokio::select! {
@@ -342,9 +340,7 @@ impl RequestReceiver {
                 },
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("RequestReceiver: Exiting");
     }
 
@@ -376,8 +372,6 @@ impl FragmentedMessageReceiver {
 
     #[cfg(not(target_arch = "wasm32"))]
     async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started FragmentedMessageReceiver with graceful shutdown support");
         while !shutdown.is_shutdown() {
             tokio::select! {
@@ -395,9 +389,7 @@ impl FragmentedMessageReceiver {
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        assert!(shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("FragmentedMessageReceiver: Exiting");
     }
 

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

@@ -8,13 +8,10 @@ use nymsphinx::anonymous_replies::{
 };
 use std::path::Path;
 
-#[derive(Debug, thiserror::Error)]
+#[derive(Debug)]
 pub enum ReplyKeyStorageError {
-    #[error("DB Read Error: {0}")]
     DbReadError(sled::Error),
-    #[error("DB Write Error: {0}")]
     DbWriteError(sled::Error),
-    #[error("DB Open Error: {0}")]
     DbOpenError(sled::Error),
 }
 

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

@@ -138,7 +138,7 @@ impl TopologyRefresherConfig {
 }
 
 pub struct TopologyRefresher {
-    validator_client: validator_client::client::ApiClient,
+    validator_client: validator_client::ApiClient,
     client_version: String,
 
     validator_api_urls: Vec<Url>,
@@ -154,9 +154,7 @@ impl TopologyRefresher {
         cfg.validator_api_urls.shuffle(&mut thread_rng());
 
         TopologyRefresher {
-            validator_client: validator_client::client::ApiClient::new(
+            validator_client: validator_client::ApiClient::new(cfg.validator_api_urls[0].clone()),
-                cfg.validator_api_urls[0].clone(),
-            ),
             client_version: cfg.client_version,
             validator_api_urls: cfg.validator_api_urls,
             topology_accessor,
@@ -311,9 +309,7 @@ impl TopologyRefresher {
                     },
                 }
             }
-            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            assert!(shutdown.is_shutdown_poll());
-                .await
-                .expect("Task stopped without shutdown called");
             log::debug!("TopologyRefresher: Exiting");
         })
     }

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

@@ -125,10 +125,6 @@ impl<T: NymConfig> Config<T> {
         self.client.gateway_endpoint.gateway_id = id.into();
     }
 
-    pub fn set_custom_validators(&mut self, validator_urls: Vec<Url>) {
-        self.client.validator_urls = validator_urls;
-    }
-
     pub fn set_custom_validator_apis(&mut self, validator_api_urls: Vec<Url>) {
         self.client.validator_api_urls = validator_api_urls;
     }
@@ -183,10 +179,6 @@ impl<T: NymConfig> Config<T> {
         self.client.ack_key_file.clone()
     }
 
-    pub fn get_validator_endpoints(&self) -> Vec<Url> {
-        self.client.validator_urls.clone()
-    }
-
     pub fn get_validator_api_endpoints(&self) -> Vec<Url> {
         self.client.validator_api_urls.clone()
     }
@@ -314,10 +306,6 @@ pub struct Client<T> {
     #[serde(default)]
     disabled_credentials_mode: bool,
 
-    /// Addresses to nymd validators via which the client can communicate with the chain.
-    #[serde(default)]
-    validator_urls: Vec<Url>,
-
     /// Addresses to APIs running on validator from which the client gets the view of the network.
     validator_api_urls: Vec<Url>,
 
@@ -366,7 +354,6 @@ impl<T: NymConfig> Default for Client<T> {
             version: env!("CARGO_PKG_VERSION").to_string(),
             id: "".to_string(),
             disabled_credentials_mode: true,
-            validator_urls: vec![],
             validator_api_urls: vec![],
             private_identity_key_file: Default::default(),
             public_identity_key_file: Default::default(),

clients/client-core/src/error.rs

@@ -26,7 +26,4 @@ pub enum ClientCoreError {
     CouldNotLoadExistingGatewayConfiguration(std::io::Error),
     #[error("The current network topology seem to be insufficient to route any packets through")]
     InsufficientNetworkTopology,
-
-    #[error("Unexpected exit")]
-    UnexpectedExit,
 }

clients/client-core/src/init.rs

@@ -31,7 +31,7 @@ pub async fn query_gateway_details(
     let validator_api = validator_servers
         .choose(&mut thread_rng())
         .ok_or(ClientCoreError::ListOfValidatorApisIsEmpty)?;
-    let validator_client = validator_client::client::ApiClient::new(validator_api.clone());
+    let validator_client = validator_client::ApiClient::new(validator_api.clone());
 
     log::trace!("Fetching list of gateways from: {}", validator_api);
     let gateways = validator_client.get_cached_gateways().await?;

clients/credential/src/commands.rs

@@ -14,9 +14,8 @@ use credential_storage::PersistentStorage;
 use credentials::coconut::bandwidth::{BandwidthVoucher, TOTAL_ATTRIBUTES};
 use credentials::coconut::utils::obtain_aggregate_signature;
 use crypto::asymmetric::{encryption, identity};
-use network_defaults::{NymNetworkDetails, VOUCHER_INFO};
+use network_defaults::VOUCHER_INFO;
 use validator_client::nymd::tx::Hash;
-use validator_client::{CoconutApiClient, Config};
 
 use crate::client::Client;
 use crate::error::{CredentialClientError, Result};
@@ -108,9 +107,10 @@ pub(crate) struct GetCredential {
     /// The hash of a successful deposit transaction
     #[clap(long)]
     tx_hash: String,
-    /// The nymd URL that should be used
+    /// The URLs to the validator-api endpoints the are run as coconut signer authorities, separated
+    /// by comma (,)
     #[clap(long)]
-    nymd_url: String,
+    signer_authorities: String,
     /// If we want to get the signature without attaching a blind sign request; it is expected that
     /// there is already a signature stored on the signer
     #[clap(long, parse(from_flag))]
@@ -124,10 +124,7 @@ impl Execute for GetCredential {
             .get::<State>(&self.tx_hash)
             .ok_or(CredentialClientError::NoDeposit)?;
 
-        let network_details = NymNetworkDetails::new_from_env();
+        let urls = config::parse_validators(&self.signer_authorities);
-        let config = Config::try_from_nym_network_details(&network_details)?;
-        let client = validator_client::Client::new_query(config)?;
-        let coconut_api_clients = CoconutApiClient::all_coconut_api_clients(&client).await?;
 
         let params = Parameters::new(TOTAL_ATTRIBUTES).unwrap();
         let bandwidth_credential_attributes = if self.__no_request {
@@ -181,12 +178,8 @@ impl Execute for GetCredential {
         )?);
         db.set(&self.tx_hash, &state).unwrap();
 
-        let signature = obtain_aggregate_signature(
+        let signature =
-            &params,
+            obtain_aggregate_signature(&params, &bandwidth_credential_attributes, &urls).await?;
-            &bandwidth_credential_attributes,
-            &coconut_api_clients,
-        )
-        .await?;
         shared_storage
             .insert_coconut_credential(
                 state.amount.to_string(),

clients/credential/src/error.rs

@@ -8,7 +8,6 @@ use credentials::error::Error as CredentialError;
 use crypto::asymmetric::encryption::KeyRecoveryError;
 use crypto::asymmetric::identity::Ed25519RecoveryError;
 use validator_client::nymd::error::NymdError;
-use validator_client::ValidatorClientError;
 
 pub type Result<T> = std::result::Result<T, CredentialClientError>;
 
@@ -17,9 +16,6 @@ pub enum CredentialClientError {
     #[error("Nymd error: {0}")]
     Nymd(#[from] NymdError),
 
-    #[error("Validator client error: {0}")]
-    ValidatorClientError(#[from] ValidatorClientError),
-
     #[error("Credential error: {0}")]
     Credential(#[from] CredentialError),
 

clients/credential/src/main.rs

@@ -22,7 +22,7 @@ cfg_if::cfg_if! {
         #[clap(author = "Nymtech", version, about)]
         struct Cli {
             /// Path pointing to an env file that configures the client.
-            #[clap(short, long)]
+            #[clap(long)]
             pub(crate) config_env_file: Option<std::path::PathBuf>,
 
             /// Path where the sqlite credental database will be located.

clients/native/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "nym-client"
-version = "1.1.0"
+version = "1.0.2"
 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.65"
+rust-version = "1.56"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -33,7 +33,6 @@ tokio-tungstenite = "0.14" # websocket
 
 ## internal
 client-core = { path = "../client-core" }
-client-connections = { path = "../../common/client-connections" }
 coconut-interface = { path = "../../common/coconut-interface", optional = true }
 config = { path = "../../common/config" }
 completions = { path = "../../common/completions" }
@@ -51,7 +50,6 @@ topology = { path = "../../common/topology" }
 validator-client = { path = "../../common/client-libs/validator-client", features = ["nymd-client"] }
 version-checker = { path = "../../common/version-checker" }
 websocket-requests = { path = "websocket-requests" }
-tap = "1.0.1"
 
 [features]
 coconut = ["coconut-interface", "credentials", "credentials/coconut", "gateway-requests/coconut", "gateway-client/coconut", "client-core/coconut"]

clients/native/examples/websocket_binarysend.rs

@@ -43,7 +43,6 @@ async fn send_file_with_reply() {
         recipient,
         message: read_data,
         with_reply_surb: true,
-        connection_id: Some(0),
     };
 
     println!("sending content of 'dummy_file' over the mix network...");
@@ -92,7 +91,6 @@ async fn send_file_without_reply() {
         recipient,
         message: read_data,
         with_reply_surb: false,
-        connection_id: Some(0),
     };
 
     println!("sending content of 'dummy_file' over the mix network...");

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

@@ -23,13 +23,6 @@ id = '{{ client.id }}'
 # to claim bandwidth without presenting bandwidth credentials.
 disabled_credentials_mode = {{ client.disabled_credentials_mode }}
 
-# Addresses to nymd validators via which the client can communicate with the chain.
-validator_urls = [
-    {{#each client.validator_urls }}
-        '{{this}}',
-    {{/each}}
-]
-
 # Addresses to APIs running on validator from which the client gets the view of the network.
 validator_api_urls = [
     {{#each client.validator_api_urls }}

clients/native/src/client/mod.rs

@@ -1,9 +1,6 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use client_connections::{
-    ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths, TransmissionLane,
-};
 use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
 use client_core::client::inbound_messages::{
     InputMessage, InputMessageReceiver, InputMessageSender,
@@ -34,7 +31,6 @@ use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::addressing::nodes::NodeIdentity;
 use nymsphinx::anonymous_replies::ReplySurb;
 use nymsphinx::receiver::ReconstructedMessage;
-use tap::TapFallible;
 use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
 
 use crate::client::config::{Config, SocketType};
@@ -114,7 +110,6 @@ impl NymClient {
         stream.start_with_shutdown(shutdown);
     }
 
-    #[allow(clippy::too_many_arguments)]
     fn start_real_traffic_controller(
         &self,
         topology_accessor: TopologyAccessor,
@@ -122,8 +117,6 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
-        lane_queue_lengths: LaneQueueLengths,
-        client_connection_rx: ConnectionCommandReceiver,
         shutdown: ShutdownListener,
     ) {
         let mut controller_config = real_messages_control::Config::new(
@@ -153,8 +146,6 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
-            lane_queue_lengths,
-            client_connection_rx,
         )
         .start_with_shutdown(shutdown);
     }
@@ -201,22 +192,11 @@ impl NymClient {
             .expect("provided gateway id is invalid!");
 
         #[cfg(feature = "coconut")]
-        let bandwidth_controller = {
+        let bandwidth_controller = BandwidthController::new(
-            let details = network_defaults::NymNetworkDetails::new_from_env();
+            credential_storage::initialise_storage(self.config.get_base().get_database_path())
-            let client_config = validator_client::Config::try_from_nym_network_details(&details)
+                .await,
-                .expect("failed to construct validator client config");
+            self.config.get_base().get_validator_api_endpoints(),
-            let client = validator_client::Client::new_query(client_config)
+        );
-                .expect("Could not construct query client");
-            let coconut_api_clients =
-                validator_client::CoconutApiClient::all_coconut_api_clients(&client)
-                    .await
-                    .expect("Could not query api clients");
-            BandwidthController::new(
-                credential_storage::initialise_storage(self.config.get_base().get_database_path())
-                    .await,
-                coconut_api_clients,
-            )
-        };
         #[cfg(not(feature = "coconut"))]
         let bandwidth_controller = BandwidthController::new(
             credential_storage::initialise_storage(self.config.get_base().get_database_path())
@@ -299,18 +279,11 @@ impl NymClient {
         &self,
         buffer_requester: ReceivedBufferRequestSender,
         msg_input: InputMessageSender,
-        shared_lane_queue_lengths: LaneQueueLengths,
-        client_connection_tx: ConnectionCommandSender,
     ) {
         info!("Starting websocket listener...");
 
-        let websocket_handler = websocket::Handler::new(
+        let websocket_handler =
-            msg_input,
+            websocket::Handler::new(msg_input, buffer_requester, self.as_mix_recipient());
-            client_connection_tx,
-            buffer_requester,
-            &self.as_mix_recipient(),
-            shared_lane_queue_lengths,
-        );
 
         websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler);
     }
@@ -318,35 +291,27 @@ impl NymClient {
     /// EXPERIMENTAL DIRECT RUST API
     /// It's untested and there are absolutely no guarantees about it (but seems to have worked
     /// well enough in local tests)
-    pub async fn send_message(
+    pub fn send_message(&mut self, recipient: Recipient, message: Vec<u8>, with_reply_surb: bool) {
-        &mut self,
+        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb);
-        recipient: Recipient,
-        message: Vec<u8>,
-        with_reply_surb: bool,
-    ) {
-        let lane = TransmissionLane::General;
-        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb, lane);
 
         self.input_tx
             .as_ref()
             .expect("start method was not called before!")
-            .send(input_msg)
+            .unbounded_send(input_msg)
-            .await
+            .unwrap();
-            .expect("InputMessageReceiver has stopped receiving!");
     }
 
     /// EXPERIMENTAL DIRECT RUST API
     /// It's untested and there are absolutely no guarantees about it (but seems to have worked
     /// well enough in local tests)
-    pub async fn send_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) {
+    pub fn send_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) {
         let input_msg = InputMessage::new_reply(reply_surb, message);
 
         self.input_tx
             .as_ref()
             .expect("start method was not called before!")
-            .send(input_msg)
+            .unbounded_send(input_msg)
-            .await
+            .unwrap();
-            .expect("InputMessageReceiver has stopped receiving!");
     }
 
     /// EXPERIMENTAL DIRECT RUST API
@@ -403,7 +368,7 @@ impl NymClient {
         let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
 
         // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
+        let (input_sender, input_receiver) = mpsc::unbounded::<InputMessage>();
 
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
@@ -411,10 +376,7 @@ impl NymClient {
 
         let reply_key_storage =
             ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
-                .tap_err(|err| {
+                .expect("Failed to load reply key storage!");
-                    log::error!("Failed to load reply key storage - is it perhaps already in use?");
-                    log::error!("{}", err);
-                })?;
 
         // Shutdown notifier for signalling tasks to stop
         let shutdown = ShutdownNotifier::default();
@@ -441,22 +403,12 @@ impl NymClient {
         let sphinx_message_sender =
             Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe());
 
-        // Channels that the websocket listener can use to signal downstream to the real traffic
-        // controller that connections are closed.
-        let (client_connection_tx, client_connection_rx) = mpsc::unbounded();
-
-        // Shared queue length data. Published by the `OutQueueController` in the client, and used
-        // primarily to throttle incoming connections (e.g socks5 for attached network-requesters)
-        let shared_lane_queue_lengths = LaneQueueLengths::new();
-
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             reply_key_storage,
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
-            shared_lane_queue_lengths.clone(),
-            client_connection_rx,
             shutdown.subscribe(),
         );
 
@@ -473,12 +425,9 @@ impl NymClient {
         }
 
         match self.config.get_socket_type() {
-            SocketType::WebSocket => self.start_websocket_listener(
+            SocketType::WebSocket => {
-                received_buffer_request_sender,
+                self.start_websocket_listener(received_buffer_request_sender, input_sender)
-                input_sender,
+            }
-                shared_lane_queue_lengths,
-                client_connection_tx,
-            ),
             SocketType::None => {
                 // if we did not start the socket, it means we're running (supposedly) in the native mode
                 // and hence we should announce 'ourselves' to the buffer

clients/native/src/commands/init.rs

@@ -25,13 +25,9 @@ pub(crate) struct Init {
     #[clap(long)]
     force_register_gateway: bool,
 
-    /// Comma separated list of rest endpoints of the nymd validators
+    /// Comma separated list of rest endpoints of the validators
     #[clap(long)]
-    nymd_validators: Option<String>,
+    validators: Option<String>,
-
-    /// Comma separated list of rest endpoints of the API validators
-    #[clap(long)]
-    api_validators: Option<String>,
 
     /// Whether to not start the websocket
     #[clap(long)]
@@ -56,8 +52,7 @@ pub(crate) struct Init {
 impl From<Init> for OverrideConfig {
     fn from(init_config: Init) -> Self {
         OverrideConfig {
-            nymd_validators: init_config.nymd_validators,
+            validators: init_config.validators,
-            api_validators: init_config.api_validators,
             disable_socket: init_config.disable_socket,
             port: init_config.port,
             fastmode: init_config.fastmode,

clients/native/src/commands/mod.rs

@@ -50,7 +50,7 @@ fn long_version_static() -> &'static str {
 #[clap(author = "Nymtech", version, long_version = long_version_static(), about)]
 pub(crate) struct Cli {
     /// Path pointing to an env file that configures the client.
-    #[clap(short, long)]
+    #[clap(long)]
     pub(crate) config_env_file: Option<std::path::PathBuf>,
 
     #[clap(subcommand)]
@@ -75,8 +75,7 @@ pub(crate) enum Commands {
 
 // Configuration that can be overridden.
 pub(crate) struct OverrideConfig {
-    nymd_validators: Option<String>,
+    validators: Option<String>,
-    api_validators: Option<String>,
     disable_socket: bool,
     port: Option<u16>,
     fastmode: bool,
@@ -99,18 +98,7 @@ pub(crate) async fn execute(args: &Cli) -> Result<(), ClientError> {
 }
 
 pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Config {
-    if let Some(raw_validators) = args.nymd_validators {
+    if let Some(raw_validators) = args.validators {
-        config
-            .get_base_mut()
-            .set_custom_validators(config::parse_validators(&raw_validators));
-    } else if std::env::var(network_defaults::var_names::CONFIGURED).is_ok() {
-        let raw_validators = std::env::var(network_defaults::var_names::NYMD_VALIDATOR)
-            .expect("nymd validator not set");
-        config
-            .get_base_mut()
-            .set_custom_validators(config::parse_validators(&raw_validators));
-    }
-    if let Some(raw_validators) = args.api_validators {
         config
             .get_base_mut()
             .set_custom_validator_apis(config::parse_validators(&raw_validators));

clients/native/src/commands/run.rs

@@ -18,13 +18,9 @@ pub(crate) struct Run {
     #[clap(long)]
     id: String,
 
-    /// Comma separated list of rest endpoints of the nymd validators
+    /// Comma separated list of rest endpoints of the validators
     #[clap(long)]
-    nymd_validators: Option<String>,
+    validators: Option<String>,
-
-    /// Comma separated list of rest endpoints of the API validators
-    #[clap(long)]
-    api_validators: Option<String>,
 
     /// Id of the gateway we want to connect to. If overridden, it is user's responsibility to
     /// ensure prior registration happened
@@ -49,8 +45,7 @@ pub(crate) struct Run {
 impl From<Run> for OverrideConfig {
     fn from(run_config: Run) -> Self {
         OverrideConfig {
-            nymd_validators: run_config.nymd_validators,
+            validators: run_config.validators,
-            api_validators: run_config.api_validators,
             disable_socket: run_config.disable_socket,
             port: run_config.port,
             fastmode: false,

clients/native/src/error.rs

@@ -1,4 +1,4 @@
-use client_core::{client::reply_key_storage::ReplyKeyStorageError, error::ClientCoreError};
+use client_core::error::ClientCoreError;
 use crypto::asymmetric::identity::Ed25519RecoveryError;
 use gateway_client::error::GatewayClientError;
 use validator_client::ValidatorClientError;
@@ -15,8 +15,6 @@ pub enum ClientError {
     ValidatorClientError(#[from] ValidatorClientError),
     #[error("client-core error: {0}")]
     ClientCoreError(#[from] ClientCoreError),
-    #[error("Reply key storage error: {0}")]
-    ReplyKeyStorageError(#[from] ReplyKeyStorageError),
 
     #[error("Failed to load config for: {0}")]
     FailedToLoadConfig(String),

clients/native/src/websocket/handler.rs

@@ -1,9 +1,6 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use client_connections::{
-    ConnectionCommand, ConnectionCommandSender, LaneQueueLengths, TransmissionLane,
-};
 use client_core::client::{
     inbound_messages::{InputMessage, InputMessageSender},
     received_buffer::{
@@ -37,12 +34,10 @@ impl Default for ReceivedResponseType {
 
 pub(crate) struct Handler {
     msg_input: InputMessageSender,
-    client_connection_tx: ConnectionCommandSender,
     buffer_requester: ReceivedBufferRequestSender,
     self_full_address: Recipient,
     socket: Option<WebSocketStream<TcpStream>>,
     received_response_type: ReceivedResponseType,
-    lane_queue_lengths: LaneQueueLengths,
 }
 
 // clone is used to use handler on a new connection, which initially is `None`
@@ -50,12 +45,10 @@ impl Clone for Handler {
     fn clone(&self) -> Self {
         Handler {
             msg_input: self.msg_input.clone(),
-            client_connection_tx: self.client_connection_tx.clone(),
             buffer_requester: self.buffer_requester.clone(),
             self_full_address: self.self_full_address,
             socket: None,
             received_response_type: Default::default(),
-            lane_queue_lengths: self.lane_queue_lengths.clone(),
         }
     }
 }
@@ -71,85 +64,38 @@ impl Drop for Handler {
 impl Handler {
     pub(crate) fn new(
         msg_input: InputMessageSender,
-        client_connection_tx: ConnectionCommandSender,
         buffer_requester: ReceivedBufferRequestSender,
-        self_full_address: &Recipient,
+        self_full_address: Recipient,
-        lane_queue_lengths: LaneQueueLengths,
     ) -> Self {
         Handler {
             msg_input,
-            client_connection_tx,
             buffer_requester,
-            self_full_address: *self_full_address,
+            self_full_address,
             socket: None,
             received_response_type: Default::default(),
-            lane_queue_lengths,
         }
     }
 
-    async fn handle_send(
+    fn handle_send(
         &mut self,
-        recipient: &Recipient,
+        recipient: Recipient,
         message: Vec<u8>,
         with_reply_surb: bool,
-        connection_id: Option<u64>,
     ) -> Option<ServerResponse> {
-        // We map the absence of a connection id as going into the general lane.
-        let lane = connection_id.map_or(TransmissionLane::General, |id| {
-            TransmissionLane::ConnectionId(id)
-        });
-
         // the ack control is now responsible for chunking, etc.
-        let input_msg = InputMessage::new_fresh(*recipient, message, with_reply_surb, lane);
+        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb);
-        self.msg_input
+        self.msg_input.unbounded_send(input_msg).unwrap();
-            .send(input_msg)
-            .await
-            .expect("InputMessageReceiver has stopped receiving!");
 
-        // Only reply back with a `LaneQueueLength` if the sender providided a connection id
+        None
-        let connection_id = match lane {
-            TransmissionLane::General
-            | TransmissionLane::Reply
-            | TransmissionLane::Retransmission
-            | TransmissionLane::Control => return None,
-            TransmissionLane::ConnectionId(id) => id,
-        };
-
-        // on receiving a send, we reply back the current lane queue length for that connection id.
-        // Note that this does _NOT_ take into account the packets that have been received but not
-        // yet reach `OutQueueControl`, so it might be a tad low.
-        let Ok(lane_queue_lengths) = self.lane_queue_lengths.lock() else {
-            log::warn!(
-                "Failed to get the lane queue length lock, \
-                not responding back with the current queue length"
-            );
-            return None;
-        };
-
-        let queue_length = lane_queue_lengths.get(&lane).unwrap_or(0);
-        Some(ServerResponse::LaneQueueLength(connection_id, queue_length))
     }
 
-    async fn handle_reply(
+    fn handle_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) -> Option<ServerResponse> {
-        &mut self,
-        reply_surb: ReplySurb,
-        message: Vec<u8>,
-    ) -> Option<ServerResponse> {
         if message.len() > ReplySurb::max_msg_len(Default::default()) {
-            return Some(
+            return Some(ServerResponse::new_error(format!("too long message to put inside a reply SURB. Received: {} bytes and maximum is {} bytes", message.len(), ReplySurb::max_msg_len(Default::default()))));
-                ServerResponse::new_error(
-                    format!(
-                        "too long message to put inside a reply SURB. Received: {} bytes and maximum is {} bytes",
-                        message.len(), ReplySurb::max_msg_len(Default::default()))
-                    )
-                );
         }
 
         let input_msg = InputMessage::new_reply(reply_surb, message);
-        self.msg_input
+        self.msg_input.unbounded_send(input_msg).unwrap();
-            .send(input_msg)
-            .await
-            .expect("InputMessageReceiver has stopped receiving!");
 
         None
     }
@@ -158,48 +104,22 @@ impl Handler {
         ServerResponse::SelfAddress(self.self_full_address)
     }
 
-    fn handle_closed_connection(&self, connection_id: u64) -> Option<ServerResponse> {
+    fn handle_request(&mut self, request: ClientRequest) -> Option<ServerResponse> {
-        self.client_connection_tx
-            .unbounded_send(ConnectionCommand::Close(connection_id))
-            .unwrap();
-        None
-    }
-
-    fn handle_get_lane_queue_length(&self, connection_id: u64) -> Option<ServerResponse> {
-        let Ok(lane_queue_lengths) = self.lane_queue_lengths.lock() else {
-            log::warn!(
-                "Failed to get the lane queue length lock, not responding back with the current queue length"
-            );
-            return None;
-        };
-
-        let lane = TransmissionLane::ConnectionId(connection_id);
-        let queue_length = lane_queue_lengths.get(&lane).unwrap_or(0);
-        Some(ServerResponse::LaneQueueLength(connection_id, queue_length))
-    }
-
-    async fn handle_request(&mut self, request: ClientRequest) -> Option<ServerResponse> {
         match request {
             ClientRequest::Send {
                 recipient,
                 message,
                 with_reply_surb,
-                connection_id,
+            } => self.handle_send(recipient, message, with_reply_surb),
-            } => {
-                self.handle_send(&recipient, message, with_reply_surb, connection_id)
-                    .await
-            }
             ClientRequest::Reply {
                 message,
                 reply_surb,
-            } => self.handle_reply(reply_surb, message).await,
+            } => self.handle_reply(reply_surb, message),
             ClientRequest::SelfAddress => Some(self.handle_self_address()),
-            ClientRequest::ClosedConnection(id) => self.handle_closed_connection(id),
-            ClientRequest::GetLaneQueueLength(id) => self.handle_get_lane_queue_length(id),
         }
     }
 
-    async fn handle_text_message(&mut self, msg: String) -> Option<WsMessage> {
+    fn handle_text_message(&mut self, msg: String) -> Option<WsMessage> {
         debug!("Handling text message request");
         trace!("Content: {:?}", msg);
 
@@ -208,13 +128,13 @@ impl Handler {
 
         let response = match client_request {
             Err(err) => Some(ServerResponse::Error(err)),
-            Ok(req) => self.handle_request(req).await,
+            Ok(req) => self.handle_request(req),
         };
 
         response.map(|resp| WsMessage::text(resp.into_text()))
     }
 
-    async fn handle_binary_message(&mut self, msg: &[u8]) -> Option<WsMessage> {
+    fn handle_binary_message(&mut self, msg: Vec<u8>) -> Option<WsMessage> {
         debug!("Handling binary message request");
 
         self.received_response_type = ReceivedResponseType::Binary;
@@ -222,23 +142,49 @@ impl Handler {
 
         let response = match client_request {
             Err(err) => Some(ServerResponse::Error(err)),
-            Ok(req) => self.handle_request(req).await,
+            Ok(req) => self.handle_request(req),
         };
 
         response.map(|resp| WsMessage::Binary(resp.into_binary()))
     }
 
-    async fn handle_ws_request(&mut self, raw_request: WsMessage) -> Option<WsMessage> {
+    fn handle_ws_request(&mut self, raw_request: WsMessage) -> Option<WsMessage> {
         // apparently tungstenite auto-handles ping/pong/close messages so for now let's ignore
         // them and let's test that claim. If that's not the case, just copy code from
         // old version of this file.
         match raw_request {
-            WsMessage::Text(text_message) => self.handle_text_message(text_message).await,
+            WsMessage::Text(text_message) => self.handle_text_message(text_message),
-            WsMessage::Binary(binary_message) => self.handle_binary_message(&binary_message).await,
+            WsMessage::Binary(binary_message) => self.handle_binary_message(binary_message),
             _ => None,
         }
     }
 
+    // I'm still not entirely sure why `send_all` requires `TryStream` rather than `Stream`, but
+    // let's just play along for now
+    fn prepare_reconstructed_binary(
+        &self,
+        reconstructed_messages: Vec<ReconstructedMessage>,
+    ) -> Vec<Result<WsMessage, WsError>> {
+        reconstructed_messages
+            .into_iter()
+            .map(ServerResponse::Received)
+            .map(|resp| Ok(WsMessage::Binary(resp.into_binary())))
+            .collect()
+    }
+
+    // I'm still not entirely sure why `send_all` requires `TryStream` rather than `Stream`, but
+    // let's just play along for now
+    fn prepare_reconstructed_text(
+        &self,
+        reconstructed_messages: Vec<ReconstructedMessage>,
+    ) -> Vec<Result<WsMessage, WsError>> {
+        reconstructed_messages
+            .into_iter()
+            .map(ServerResponse::Received)
+            .map(|resp| Ok(WsMessage::Text(resp.into_text())))
+            .collect()
+    }
+
     async fn push_websocket_received_plaintexts(
         &mut self,
         reconstructed_messages: Vec<ReconstructedMessage>,
@@ -247,8 +193,10 @@ impl Handler {
         // if it's text or binary, but for time being we use the naive assumption that if
         // client is sending Message::Text it expects text back. Same for Message::Binary
         let response_messages = match self.received_response_type {
-            ReceivedResponseType::Binary => prepare_reconstructed_binary(reconstructed_messages),
+            ReceivedResponseType::Binary => {
-            ReceivedResponseType::Text => prepare_reconstructed_text(reconstructed_messages),
+                self.prepare_reconstructed_binary(reconstructed_messages)
+            }
+            ReceivedResponseType::Text => self.prepare_reconstructed_text(reconstructed_messages),
         };
 
         let mut send_stream = futures::stream::iter(response_messages);
@@ -296,7 +244,7 @@ impl Handler {
                         break;
                     }
 
-                    if let Some(response) = self.handle_ws_request(socket_msg).await {
+                    if let Some(response) = self.handle_ws_request(socket_msg) {
                         if let Err(err) = self.send_websocket_response(response).await {
                             warn!(
                                 "Failed to send message over websocket: {}. Assuming the connection is dead.",
@@ -343,27 +291,3 @@ impl Handler {
         self.listen_for_requests(reconstructed_receiver).await;
     }
 }
-
-// I'm still not entirely sure why `send_all` requires `TryStream` rather than `Stream`, but
-// let's just play along for now
-fn prepare_reconstructed_binary(
-    reconstructed_messages: Vec<ReconstructedMessage>,
-) -> Vec<Result<WsMessage, WsError>> {
-    reconstructed_messages
-        .into_iter()
-        .map(ServerResponse::Received)
-        .map(|resp| Ok(WsMessage::Binary(resp.into_binary())))
-        .collect()
-}
-
-// I'm still not entirely sure why `send_all` requires `TryStream` rather than `Stream`, but
-// let's just play along for now
-fn prepare_reconstructed_text(
-    reconstructed_messages: Vec<ReconstructedMessage>,
-) -> Vec<Result<WsMessage, WsError>> {
-    reconstructed_messages
-        .into_iter()
-        .map(ServerResponse::Received)
-        .map(|resp| Ok(WsMessage::Text(resp.into_text())))
-        .collect()
-}

clients/native/websocket-requests/src/requests.rs

@@ -20,12 +20,6 @@ pub const REPLY_REQUEST_TAG: u8 = 0x01;
 /// Value tag representing [`SelfAddress`] variant of the [`ClientRequest`]
 pub const SELF_ADDRESS_REQUEST_TAG: u8 = 0x02;
 
-/// Value tag representing [`ClosedConnection`] variant of the [`ClientRequest`]
-pub const CLOSED_CONNECTION_REQUEST_TAG: u8 = 0x03;
-
-/// Value tag representing [`GetLaneQueueLength`] variant of the [`ClientRequest`]
-pub const GET_LANE_QUEUE_LENGHT_TAG: u8 = 0x04;
-
 #[allow(non_snake_case)]
 #[derive(Debug)]
 pub enum ClientRequest {
@@ -34,42 +28,32 @@ pub enum ClientRequest {
         message: Vec<u8>,
         // Perhaps we could change it to a number to indicate how many reply_SURBs we want to include?
         with_reply_surb: bool,
-        connection_id: Option<u64>,
     },
     Reply {
         message: Vec<u8>,
         reply_surb: ReplySurb,
     },
     SelfAddress,
-    ClosedConnection(u64),
-    GetLaneQueueLength(u64),
 }
 
 // we could have been parsing it directly TryFrom<WsMessage>, but we want to retain
 // information about whether it came from binary or text to send appropriate response back
 impl ClientRequest {
-    // SEND_REQUEST_TAG || with_surb || recipient || conn_id || data_len || data
+    // SEND_REQUEST_TAG || with_surb || recipient || data_len || data
-    fn serialize_send(
+    fn serialize_send(recipient: Recipient, data: Vec<u8>, with_reply_surb: bool) -> Vec<u8> {
-        recipient: Recipient,
-        data: Vec<u8>,
-        with_reply_surb: bool,
-        connection_id: Option<u64>,
-    ) -> Vec<u8> {
         let data_len_bytes = (data.len() as u64).to_be_bytes();
-        let conn_id_bytes = connection_id.unwrap_or(0).to_be_bytes();
         std::iter::once(SEND_REQUEST_TAG)
             .chain(std::iter::once(with_reply_surb as u8))
             .chain(recipient.to_bytes().iter().cloned()) // will not be length prefixed because the length is constant
-            .chain(conn_id_bytes.iter().cloned())
             .chain(data_len_bytes.iter().cloned())
             .chain(data.into_iter())
             .collect()
     }
 
-    // SEND_REQUEST_TAG || with_reply || recipient || conn_id || data_len || data
+    // SEND_REQUEST_TAG || with_reply || recipient || data_len || data
     fn deserialize_send(b: &[u8]) -> Result<Self, error::Error> {
-        // we need to have at least 1 (tag) + 1 (reply flag) + Recipient::LEN + 2*sizeof<u64> bytes
+        // we need to have at least 1 (tag) + 1 (reply flag) + Recipient::LEN + sizeof<u64> bytes
-        if b.len() < 2 + Recipient::LEN + 2 * size_of::<u64>() {
+        if b.len() < 2 + Recipient::LEN + size_of::<u64>() {
             return Err(error::Error::new(
                 ErrorKind::TooShortRequest,
                 "not enough data provided to recover 'send'".to_string(),
@@ -102,20 +86,9 @@ impl ClientRequest {
             }
         };
 
-        let mut connection_id_bytes = [0u8; size_of::<u64>()];
+        let data_len_bytes = &b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::<u64>()];
-        connection_id_bytes
-            .copy_from_slice(&b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::<u64>()]);
-        let connection_id = u64::from_be_bytes(connection_id_bytes);
-        let connection_id = if connection_id == 0 {
-            None
-        } else {
-            Some(connection_id)
-        };
-
-        let data_len_bytes =
-            &b[2 + Recipient::LEN + size_of::<u64>()..2 + Recipient::LEN + 2 * size_of::<u64>()];
         let data_len = u64::from_be_bytes(data_len_bytes.try_into().unwrap());
-        let data = &b[2 + Recipient::LEN + 2 * size_of::<u64>()..];
+        let data = &b[2 + Recipient::LEN + size_of::<u64>()..];
         if data.len() as u64 != data_len {
             return Err(error::Error::new(
                 ErrorKind::MalformedRequest,
@@ -131,12 +104,11 @@ impl ClientRequest {
             with_reply_surb,
             recipient,
             message: data.to_vec(),
-            connection_id,
         })
     }
 
     // REPLY_REQUEST_TAG || surb_len || surb || message_len || message
-    fn serialize_reply(message: Vec<u8>, reply_surb: &ReplySurb) -> Vec<u8> {
+    fn serialize_reply(message: Vec<u8>, reply_surb: ReplySurb) -> Vec<u8> {
         let reply_surb_bytes = reply_surb.to_bytes();
         let surb_len_bytes = (reply_surb_bytes.len() as u64).to_be_bytes();
         let message_len_bytes = (message.len() as u64).to_be_bytes();
@@ -230,79 +202,20 @@ impl ClientRequest {
         ClientRequest::SelfAddress
     }
 
-    // CLOSED_CONNECTION_REQUEST_TAG
-    fn serialize_closed_connection(connection_id: u64) -> Vec<u8> {
-        let conn_id_bytes = connection_id.to_be_bytes();
-        std::iter::once(CLOSED_CONNECTION_REQUEST_TAG)
-            .chain(conn_id_bytes.iter().copied())
-            .collect()
-    }
-
-    // CLOSED_CONNECTION_REQUEST_TAG
-    fn deserialize_closed_connection(b: &[u8]) -> Result<Self, error::Error> {
-        if b.len() != 1 + size_of::<u64>() {
-            return Err(error::Error::new(
-                ErrorKind::MalformedRequest,
-                "the received closed connection has invalid length".to_string(),
-            ));
-        }
-
-        // this MUST match because it was called by 'deserialize'
-        debug_assert_eq!(b[0], CLOSED_CONNECTION_REQUEST_TAG);
-
-        let mut connection_id_bytes = [0u8; size_of::<u64>()];
-        connection_id_bytes.copy_from_slice(&b[1..=size_of::<u64>()]);
-        let connection_id = u64::from_be_bytes(connection_id_bytes);
-
-        Ok(ClientRequest::ClosedConnection(connection_id))
-    }
-
-    // GET_LANE_QUEUE_LENGHT_TAG
-    fn serialize_get_lane_queue_lengths(connection_id: u64) -> Vec<u8> {
-        let conn_id_bytes = connection_id.to_be_bytes();
-        std::iter::once(GET_LANE_QUEUE_LENGHT_TAG)
-            .chain(conn_id_bytes.iter().copied())
-            .collect()
-    }
-
-    // GET_LANE_QUEUE_LENGHT_TAG
-    fn deserialize_get_lane_queue_length(b: &[u8]) -> Result<Self, error::Error> {
-        if b.len() != 1 + size_of::<u64>() {
-            return Err(error::Error::new(
-                ErrorKind::MalformedRequest,
-                "the received get lane queue length has invalid length".to_string(),
-            ));
-        }
-
-        // this MUST match because it was called by 'deserialize'
-        debug_assert_eq!(b[0], GET_LANE_QUEUE_LENGHT_TAG);
-
-        let mut connection_id_bytes = [0u8; size_of::<u64>()];
-        connection_id_bytes.copy_from_slice(&b[1..=size_of::<u64>()]);
-        let connection_id = u64::from_be_bytes(connection_id_bytes);
-
-        Ok(ClientRequest::GetLaneQueueLength(connection_id))
-    }
-
     pub fn serialize(self) -> Vec<u8> {
         match self {
             ClientRequest::Send {
                 recipient,
                 message,
                 with_reply_surb,
-                connection_id,
+            } => Self::serialize_send(recipient, message, with_reply_surb),
-            } => Self::serialize_send(recipient, message, with_reply_surb, connection_id),
 
             ClientRequest::Reply {
                 message,
                 reply_surb,
-            } => Self::serialize_reply(message, &reply_surb),
+            } => Self::serialize_reply(message, reply_surb),
 
             ClientRequest::SelfAddress => Self::serialize_self_address(),
-
-            ClientRequest::ClosedConnection(id) => Self::serialize_closed_connection(id),
-
-            ClientRequest::GetLaneQueueLength(id) => Self::serialize_get_lane_queue_lengths(id),
         }
     }
 
@@ -332,17 +245,15 @@ impl ClientRequest {
             SEND_REQUEST_TAG => Self::deserialize_send(b),
             REPLY_REQUEST_TAG => Self::deserialize_reply(b),
             SELF_ADDRESS_REQUEST_TAG => Ok(Self::deserialize_self_address(b)),
-            CLOSED_CONNECTION_REQUEST_TAG => Self::deserialize_closed_connection(b),
-            GET_LANE_QUEUE_LENGHT_TAG => Self::deserialize_get_lane_queue_length(b),
             n => Err(error::Error::new(
                 ErrorKind::UnknownRequest,
-                format!("type {n}"),
+                format!("type {}", n),
             )),
         }
     }
 
-    pub fn try_from_binary(raw_req: &[u8]) -> Result<Self, error::Error> {
+    pub fn try_from_binary(raw_req: Vec<u8>) -> Result<Self, error::Error> {
-        Self::deserialize(raw_req)
+        Self::deserialize(&raw_req)
     }
 
     pub fn try_from_text(raw_req: String) -> Result<Self, error::Error> {
@@ -369,7 +280,6 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: false,
-            connection_id: Some(42),
         };
 
         let bytes = send_request_no_surb.serialize();
@@ -379,12 +289,10 @@ mod tests {
                 recipient,
                 message,
                 with_reply_surb,
-                connection_id,
             } => {
                 assert_eq!(recipient.to_string(), recipient_string);
                 assert_eq!(message, b"foomp".to_vec());
-                assert!(!with_reply_surb);
+                assert!(!with_reply_surb)
-                assert_eq!(connection_id, Some(42))
             }
             _ => unreachable!(),
         }
@@ -393,7 +301,6 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: true,
-            connection_id: None,
         };
 
         let bytes = send_request_surb.serialize();
@@ -403,12 +310,10 @@ mod tests {
                 recipient,
                 message,
                 with_reply_surb,
-                connection_id,
             } => {
                 assert_eq!(recipient.to_string(), recipient_string);
                 assert_eq!(message, b"foomp".to_vec());
-                assert!(with_reply_surb);
+                assert!(with_reply_surb)
-                assert_eq!(connection_id, None)
             }
             _ => unreachable!(),
         }
@@ -447,15 +352,4 @@ mod tests {
             _ => unreachable!(),
         }
     }
-
-    #[test]
-    fn close_connection_request_serialization_works() {
-        let close_connection_request = ClientRequest::ClosedConnection(42);
-        let bytes = close_connection_request.serialize();
-        let recovered = ClientRequest::deserialize(&bytes).unwrap();
-        match recovered {
-            ClientRequest::ClosedConnection(id) => assert_eq!(id, 42),
-            _ => unreachable!(),
-        }
-    }
 }

clients/native/websocket-requests/src/responses.rs

@@ -23,14 +23,10 @@ pub const RECEIVED_RESPONSE_TAG: u8 = 0x01;
 /// Value tag representing [`SelfAddress`] variant of the [`ServerResponse`]
 pub const SELF_ADDRESS_RESPONSE_TAG: u8 = 0x02;
 
-/// Value tag representing [`LaneQueueLength`] variant of the [`ServerResponse`]
-pub const LANE_QUEUE_LENGTH_RESPONSE_TAG: u8 = 0x03;
-
 #[derive(Debug)]
 pub enum ServerResponse {
     Received(ReconstructedMessage),
     SelfAddress(Recipient),
-    LaneQueueLength(u64, usize),
     Error(error::Error),
 }
 
@@ -197,31 +193,6 @@ impl ServerResponse {
         Ok(ServerResponse::SelfAddress(recipient))
     }
 
-    // LANE_QUEUE_LENGTH_RESPONSE_TAG || lane || queue_length
-    fn serialize_lane_queue_length(lane: u64, queue_length: usize) -> Vec<u8> {
-        std::iter::once(LANE_QUEUE_LENGTH_RESPONSE_TAG)
-            .chain(lane.to_be_bytes().iter().cloned())
-            .chain(queue_length.to_be_bytes().iter().cloned())
-            .collect()
-    }
-
-    // LANE_QUEUE_LENGTH_RESPONSE_TAG || lane || queue_length
-    fn deserialize_lane_queue_length(b: &[u8]) -> Result<Self, error::Error> {
-        // this MUST match because it was called by 'deserialize'
-        debug_assert_eq!(b[0], LANE_QUEUE_LENGTH_RESPONSE_TAG);
-
-        let mut lane_bytes = [0u8; size_of::<u64>()];
-        lane_bytes.copy_from_slice(&b[1..=size_of::<u64>()]);
-        let lane = u64::from_be_bytes(lane_bytes);
-
-        let mut queue_length_bytes = [0u8; size_of::<usize>()];
-        queue_length_bytes
-            .copy_from_slice(&b[1 + size_of::<u64>()..1 + size_of::<u64>() + size_of::<usize>()]);
-        let queue_length = usize::from_be_bytes(queue_length_bytes);
-
-        Ok(ServerResponse::LaneQueueLength(lane, queue_length))
-    }
-
     // ERROR_RESPONSE_TAG || err_code || msg_len || msg
     fn serialize_error(error: error::Error) -> Vec<u8> {
         let message_len_bytes = (error.message.len() as u64).to_be_bytes();
@@ -301,9 +272,6 @@ impl ServerResponse {
                 Self::serialize_received(reconstructed_message)
             }
             ServerResponse::SelfAddress(address) => Self::serialize_self_address(address),
-            ServerResponse::LaneQueueLength(lane, queue_length) => {
-                Self::serialize_lane_queue_length(lane, queue_length)
-            }
             ServerResponse::Error(err) => Self::serialize_error(err),
         }
     }
@@ -334,7 +302,6 @@ impl ServerResponse {
         match response_tag {
             RECEIVED_RESPONSE_TAG => Self::deserialize_received(b),
             SELF_ADDRESS_RESPONSE_TAG => Self::deserialize_self_address(b),
-            LANE_QUEUE_LENGTH_RESPONSE_TAG => Self::deserialize_lane_queue_length(b),
             ERROR_RESPONSE_TAG => Self::deserialize_error(b),
             n => Err(error::Error::new(
                 ErrorKind::UnknownResponse,
@@ -411,20 +378,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn lane_queue_length_response_serialization_works() {
-        let lane_queue_length_response = ServerResponse::LaneQueueLength(13, 42);
-        let bytes = lane_queue_length_response.serialize();
-        let recovered = ServerResponse::deserialize(&bytes).unwrap();
-        match recovered {
-            ServerResponse::LaneQueueLength(lane, queue_length) => {
-                assert_eq!(lane, 13);
-                assert_eq!(queue_length, 42)
-            }
-            _ => unreachable!(),
-        }
-    }
-
     #[test]
     fn error_response_serialization_works() {
         let dummy_error = error::Error::new(ErrorKind::UnknownRequest, "foomp message".to_string());

clients/native/websocket-requests/src/text.rs

@@ -20,7 +20,6 @@ pub(super) enum ClientRequestText {
         message: String,
         recipient: String,
         with_reply_surb: bool,
-        connection_id: Option<u64>,
     },
     SelfAddress,
     #[serde(rename_all = "camelCase")]
@@ -47,7 +46,6 @@ impl TryInto<ClientRequest> for ClientRequestText {
                 message,
                 recipient,
                 with_reply_surb,
-                connection_id,
             } => {
                 let message_bytes = message.into_bytes();
                 let recipient = Recipient::try_from_base58_string(recipient).map_err(|err| {
@@ -58,7 +56,6 @@ impl TryInto<ClientRequest> for ClientRequestText {
                     message: message_bytes,
                     recipient,
                     with_reply_surb,
-                    connection_id,
                 })
             }
             ClientRequestText::SelfAddress => Ok(ClientRequest::SelfAddress),
@@ -94,10 +91,6 @@ pub(super) enum ServerResponseText {
     SelfAddress {
         address: String,
     },
-    LaneQueueLength {
-        lane: u64,
-        queue_length: usize,
-    },
     Error {
         message: String,
     },
@@ -139,9 +132,6 @@ impl From<ServerResponse> for ServerResponseText {
             ServerResponse::SelfAddress(recipient) => ServerResponseText::SelfAddress {
                 address: recipient.to_string(),
             },
-            ServerResponse::LaneQueueLength(lane, queue_length) => {
-                ServerResponseText::LaneQueueLength { lane, queue_length }
-            }
             ServerResponse::Error(err) => ServerResponseText::Error {
                 message: err.to_string(),
             },

clients/socks5/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-socks5-client"
-version = "1.1.0"
+version = "1.0.2"
 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"
@@ -26,7 +26,6 @@ url = "2.2"
 
 # internal
 client-core = { path = "../client-core" }
-client-connections = { path = "../../common/client-connections" }
 coconut-interface = { path = "../../common/coconut-interface", optional = true }
 config = { path = "../../common/config" }
 completions = { path = "../../common/completions" }
@@ -46,7 +45,6 @@ task = { path = "../../common/task" }
 topology = { path = "../../common/topology" }
 validator-client = { path = "../../common/client-libs/validator-client", features = ["nymd-client"] }
 version-checker = { path = "../../common/version-checker" }
-tap = "1.0.1"
 
 [features]
 coconut = ["coconut-interface", "credentials", "gateway-requests/coconut", "gateway-client/coconut", "credentials/coconut", "client-core/coconut"]

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

@@ -23,13 +23,6 @@ id = '{{ client.id }}'
 # to claim bandwidth without presenting bandwidth credentials.
 disabled_credentials_mode = {{ client.disabled_credentials_mode }}
 
-# Addresses to nymd validators via which the client can communicate with the chain.
-validator_urls = [
-    {{#each client.validator_urls }}
-        '{{this}}',
-    {{/each}}
-]
-
 # Addresses to APIs running on validator from which the client gets the view of the network.
 validator_api_urls = [
     {{#each client.validator_api_urls }}

clients/socks5/src/client/mod.rs

@@ -1,7 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::error::Error;
+use std::sync::atomic::Ordering;
 
 use crate::client::config::Config;
 use crate::error::Socks5ClientError;
@@ -9,7 +9,6 @@ use crate::socks::{
     authentication::{AuthenticationMethods, Authenticator, User},
     server::SphinxSocksServer,
 };
-use client_connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
 use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
 use client_core::client::inbound_messages::{
     InputMessage, InputMessageReceiver, InputMessageSender,
@@ -37,8 +36,7 @@ use gateway_client::{
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::addressing::nodes::NodeIdentity;
-use tap::TapFallible;
+use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
-use task::{wait_for_signal_and_error, ShutdownListener, ShutdownNotifier};
 
 pub mod config;
 
@@ -112,7 +110,6 @@ impl NymClient {
         stream.start_with_shutdown(shutdown);
     }
 
-    #[allow(clippy::too_many_arguments)]
     fn start_real_traffic_controller(
         &self,
         topology_accessor: TopologyAccessor,
@@ -120,8 +117,6 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
-        client_connection_rx: ConnectionCommandReceiver,
-        lane_queue_lengths: LaneQueueLengths,
         shutdown: ShutdownListener,
     ) {
         let mut controller_config = client_core::client::real_messages_control::Config::new(
@@ -151,8 +146,6 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
-            lane_queue_lengths,
-            client_connection_rx,
         )
         .start_with_shutdown(shutdown);
     }
@@ -199,22 +192,11 @@ impl NymClient {
             .expect("provided gateway id is invalid!");
 
         #[cfg(feature = "coconut")]
-        let bandwidth_controller = {
+        let bandwidth_controller = BandwidthController::new(
-            let details = network_defaults::NymNetworkDetails::new_from_env();
+            credential_storage::initialise_storage(self.config.get_base().get_database_path())
-            let client_config = validator_client::Config::try_from_nym_network_details(&details)
+                .await,
-                .expect("failed to construct validator client config");
+            self.config.get_base().get_validator_api_endpoints(),
-            let client = validator_client::Client::new_query(client_config)
+        );
-                .expect("Could not construct query client");
-            let coconut_api_clients =
-                validator_client::CoconutApiClient::all_coconut_api_clients(&client)
-                    .await
-                    .expect("Could not query api clients");
-            BandwidthController::new(
-                credential_storage::initialise_storage(self.config.get_base().get_database_path())
-                    .await,
-                coconut_api_clients,
-            )
-        };
         #[cfg(not(feature = "coconut"))]
         let bandwidth_controller = BandwidthController::new(
             credential_storage::initialise_storage(self.config.get_base().get_database_path())
@@ -297,9 +279,7 @@ impl NymClient {
         &self,
         buffer_requester: ReceivedBufferRequestSender,
         msg_input: InputMessageSender,
-        client_connection_tx: ConnectionCommandSender,
+        shutdown: ShutdownListener,
-        lane_queue_lengths: LaneQueueLengths,
-        mut shutdown: ShutdownListener,
     ) {
         info!("Starting socks5 listener...");
         let auth_methods = vec![AuthenticationMethods::NoAuth as u8];
@@ -311,58 +291,34 @@ impl NymClient {
             authenticator,
             self.config.get_provider_mix_address(),
             self.as_mix_recipient(),
-            lane_queue_lengths,
+            shutdown,
-            shutdown.clone(),
         );
-        tokio::spawn(async move {
+        tokio::spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
-            // Ideally we should have a fully fledged task manager to check for errors in all
-            // tasks.
-            // However, pragmatically, we start out by at least reporting errors for some of the
-            // tasks that interact with the outside world and can fail in normal operation, such as
-            // network issues.
-            // TODO: replace this by a generic solution, such as a task manager that stores all
-            // JoinHandles of all spawned tasks.
-            if let Err(res) = sphinx_socks
-                .serve(msg_input, buffer_requester, client_connection_tx)
-                .await
-            {
-                shutdown.send_we_stopped(Box::new(res));
-            }
-        });
     }
 
     /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
-    pub async fn run_forever(&mut self) -> Result<(), Box<dyn Error + Send>> {
+    pub async fn run_forever(&mut self) -> Result<(), Socks5ClientError> {
-        let mut shutdown = self
+        let mut shutdown = self.start().await?;
-            .start()
+        wait_for_signal().await;
-            .await
-            .map_err(|err| Box::new(err) as Box<dyn Error + Send>)?;
-
-        let res = wait_for_signal_and_error(&mut shutdown).await;
 
         log::info!("Sending shutdown");
+        client_core::client::SHUTDOWN_HAS_BEEN_SIGNALLED.store(true, Ordering::Relaxed);
         shutdown.signal_shutdown().ok();
 
         log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
         shutdown.wait_for_shutdown().await;
 
         log::info!("Stopping nym-socks5-client");
-        res
+        Ok(())
     }
 
     // Variant of `run_forever` that listends for remote control messages
     pub async fn run_and_listen(
         &mut self,
         mut receiver: Socks5ControlMessageReceiver,
-    ) -> Result<(), Box<dyn Error + Send>> {
+    ) -> Result<(), Socks5ClientError> {
-        // Start the main task
+        let mut shutdown = self.start().await?;
-        let mut shutdown = self
+        tokio::select! {
-            .start()
-            .await
-            .map_err(|err| Box::new(err) as Box<dyn Error + Send>)?;
-
-        let res = tokio::select! {
-            biased;
             message = receiver.next() => {
                 log::debug!("Received message: {:?}", message);
                 match message {
@@ -373,26 +329,21 @@ impl NymClient {
                         log::info!("Channel closed, stopping");
                     }
                 }
-                Ok(())
-            }
-            Some(msg) = shutdown.wait_for_error() => {
-                log::info!("Task error: {:?}", msg);
-                Err(msg)
             }
             _ = tokio::signal::ctrl_c() => {
                 log::info!("Received SIGINT");
-                Ok(())
             },
-        };
+        }
 
         log::info!("Sending shutdown");
+        client_core::client::SHUTDOWN_HAS_BEEN_SIGNALLED.store(true, Ordering::Relaxed);
         shutdown.signal_shutdown().ok();
 
         log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
         shutdown.wait_for_shutdown().await;
 
         log::info!("Stopping nym-socks5-client");
-        res
+        Ok(())
     }
 
     pub async fn start(&mut self) -> Result<ShutdownNotifier, Socks5ClientError> {
@@ -410,7 +361,7 @@ impl NymClient {
         let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
 
         // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
+        let (input_sender, input_receiver) = mpsc::unbounded::<InputMessage>();
 
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
@@ -418,10 +369,7 @@ impl NymClient {
 
         let reply_key_storage =
             ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
-                .tap_err(|err| {
+                .expect("Failed to load reply key storage!");
-                    log::error!("Failed to load reply key storage - is it perhaps already in use?");
-                    log::error!("{}", err);
-                })?;
 
         // Shutdown notifier for signalling tasks to stop
         let shutdown = ShutdownNotifier::default();
@@ -448,22 +396,12 @@ impl NymClient {
         let sphinx_message_sender =
             Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe());
 
-        // Channel for announcing closed (socks5) connections by the controller.
-        // This will be forwarded to `OutQueueControl`
-        let (client_connection_tx, client_connection_rx) = mpsc::unbounded();
-
-        // Shared queue length data. Published by the `OutQueueController` in the client, and used
-        // primarily to throttle incoming connections
-        let shared_lane_queue_lengths = LaneQueueLengths::new();
-
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             reply_key_storage,
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
-            client_connection_rx,
-            shared_lane_queue_lengths.clone(),
             shutdown.subscribe(),
         );
 
@@ -482,8 +420,6 @@ impl NymClient {
         self.start_socks5_listener(
             received_buffer_request_sender,
             input_sender,
-            client_connection_tx,
-            shared_lane_queue_lengths,
             shutdown.subscribe(),
         );
 

clients/socks5/src/commands/init.rs

@@ -29,13 +29,9 @@ pub(crate) struct Init {
     #[clap(long)]
     force_register_gateway: bool,
 
-    /// Comma separated list of rest endpoints of the nymd validators
+    /// Comma separated list of rest endpoints of the validators
     #[clap(long)]
-    nymd_validators: Option<String>,
+    validators: Option<String>,
-
-    /// Comma separated list of rest endpoints of the API validators
-    #[clap(long)]
-    api_validators: Option<String>,
 
     /// Port for the socket to listen on in all subsequent runs
     #[clap(short, long)]
@@ -56,8 +52,7 @@ pub(crate) struct Init {
 impl From<Init> for OverrideConfig {
     fn from(init_config: Init) -> Self {
         OverrideConfig {
-            nymd_validators: init_config.nymd_validators,
+            validators: init_config.validators,
-            api_validators: init_config.api_validators,
             port: init_config.port,
             fastmode: init_config.fastmode,
             #[cfg(feature = "coconut")]

clients/socks5/src/commands/mod.rs

@@ -1,9 +1,8 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::error::Error;
-
 use crate::client::config::Config;
+use crate::error::Socks5ClientError;
 use clap::CommandFactory;
 use clap::{Parser, Subcommand};
 use completions::{fig_generate, ArgShell};
@@ -52,7 +51,7 @@ fn long_version_static() -> &'static str {
 #[clap(author = "Nymtech", version, long_version = long_version_static(), about)]
 pub(crate) struct Cli {
     /// Path pointing to an env file that configures the client.
-    #[clap(short, long)]
+    #[clap(long)]
     pub(crate) config_env_file: Option<std::path::PathBuf>,
 
     #[clap(subcommand)]
@@ -63,10 +62,8 @@ pub(crate) struct Cli {
 pub(crate) enum Commands {
     /// Initialise a Nym client. Do this first!
     Init(init::Init),
-
     /// Run the Nym client with provided configuration client optionally overriding set parameters
     Run(run::Run),
-
     /// Try to upgrade the client
     Upgrade(upgrade::Upgrade),
 
@@ -79,8 +76,7 @@ pub(crate) enum Commands {
 
 // Configuration that can be overridden.
 pub(crate) struct OverrideConfig {
-    nymd_validators: Option<String>,
+    validators: Option<String>,
-    api_validators: Option<String>,
     port: Option<u16>,
     fastmode: bool,
 
@@ -88,7 +84,7 @@ pub(crate) struct OverrideConfig {
     enabled_credentials_mode: bool,
 }
 
-pub(crate) async fn execute(args: &Cli) -> Result<(), Box<dyn Error + Send>> {
+pub(crate) async fn execute(args: &Cli) -> Result<(), Socks5ClientError> {
     let bin_name = "nym-socks5-client";
 
     match &args.command {
@@ -102,16 +98,7 @@ pub(crate) async fn execute(args: &Cli) -> Result<(), Box<dyn Error + Send>> {
 }
 
 pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Config {
-    if let Some(raw_validators) = args.nymd_validators {
+    if let Some(raw_validators) = args.validators {
-        config
-            .get_base_mut()
-            .set_custom_validators(parse_validators(&raw_validators));
-    } else if let Ok(raw_validators) = std::env::var(network_defaults::var_names::NYMD_VALIDATOR) {
-        config
-            .get_base_mut()
-            .set_custom_validators(parse_validators(&raw_validators));
-    }
-    if let Some(raw_validators) = args.api_validators {
         config
             .get_base_mut()
             .set_custom_validator_apis(parse_validators(&raw_validators));

clients/socks5/src/commands/run.rs

@@ -31,13 +31,9 @@ pub(crate) struct Run {
     #[clap(long)]
     gateway: Option<String>,
 
-    /// Comma separated list of rest endpoints of the nymd validators
+    /// Comma separated list of rest endpoints of the validators
     #[clap(long)]
-    nymd_validators: Option<String>,
+    validators: Option<String>,
-
-    /// Comma separated list of rest endpoints of the API validators
-    #[clap(long)]
-    api_validators: Option<String>,
 
     /// Port for the socket to listen on
     #[clap(short, long)]
@@ -53,8 +49,7 @@ pub(crate) struct Run {
 impl From<Run> for OverrideConfig {
     fn from(run_config: Run) -> Self {
         OverrideConfig {
-            nymd_validators: run_config.nymd_validators,
+            validators: run_config.validators,
-            api_validators: run_config.api_validators,
             port: run_config.port,
             fastmode: false,
 
@@ -86,16 +81,14 @@ fn version_check(cfg: &Config) -> bool {
     }
 }
 
-pub(crate) async fn execute(args: &Run) -> Result<(), Box<dyn std::error::Error + Send>> {
+pub(crate) async fn execute(args: &Run) -> Result<(), Socks5ClientError> {
     let id = &args.id;
 
     let mut config = match Config::load_from_file(Some(id)) {
         Ok(cfg) => cfg,
         Err(err) => {
             error!("Failed to load config for {}. Are you sure you have run `init` before? (Error was: {})", id, err);
-            return Err(Box::new(Socks5ClientError::FailedToLoadConfig(
+            return Err(Socks5ClientError::FailedToLoadConfig(id.to_string()));
-                id.to_string(),
-            )));
         }
     };
 
@@ -104,7 +97,7 @@ pub(crate) async fn execute(args: &Run) -> Result<(), Box<dyn std::error::Error
 
     if !version_check(&config) {
         error!("failed the local version check");
-        return Err(Box::new(Socks5ClientError::FailedLocalVersionCheck));
+        return Err(Socks5ClientError::FailedLocalVersionCheck);
     }
 
     NymClient::new(config).run_forever().await

clients/socks5/src/error.rs

@@ -1,10 +1,8 @@
-use client_core::{client::reply_key_storage::ReplyKeyStorageError, error::ClientCoreError};
+use client_core::error::ClientCoreError;
 use crypto::asymmetric::identity::Ed25519RecoveryError;
 use gateway_client::error::GatewayClientError;
 use validator_client::ValidatorClientError;
 
-use crate::socks::types::SocksProxyError;
-
 #[derive(thiserror::Error, Debug)]
 pub enum Socks5ClientError {
     #[error("I/O error: {0}")]
@@ -17,16 +15,9 @@ pub enum Socks5ClientError {
     ValidatorClientError(#[from] ValidatorClientError),
     #[error("client-core error: {0}")]
     ClientCoreError(#[from] ClientCoreError),
-    #[error("Reply key storage error: {0}")]
-    ReplyKeyStorageError(#[from] ReplyKeyStorageError),
-
-    #[error("SOCKS proxy error")]
-    SocksProxyError(SocksProxyError),
 
     #[error("Failed to load config for: {0}")]
     FailedToLoadConfig(String),
     #[error("Failed local version check, client and config mismatch")]
     FailedLocalVersionCheck,
-    #[error("Fail to bind address")]
-    FailToBindAddress,
 }

clients/socks5/src/main.rs

@@ -1,9 +1,8 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::error::Error;
-
 use clap::{crate_version, Parser};
+use error::Socks5ClientError;
 use logging::setup_logging;
 use network_defaults::setup_env;
 
@@ -13,7 +12,7 @@ pub mod error;
 pub mod socks;
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send>> {
+async fn main() -> Result<(), Socks5ClientError> {
     setup_logging();
     println!("{}", banner());
 

clients/socks5/src/socks/client.rs

@@ -4,8 +4,8 @@ use super::authentication::{AuthenticationMethods, Authenticator, User};
 use super::request::{SocksCommand, SocksRequest};
 use super::types::{ResponseCode, SocksProxyError};
 use super::{RESERVED, SOCKS_VERSION};
-use client_connections::{LaneQueueLengths, TransmissionLane};
+use client_core::client::inbound_messages::InputMessage;
-use client_core::client::inbound_messages::{InputMessage, InputMessageSender};
+use client_core::client::inbound_messages::InputMessageSender;
 use futures::channel::mpsc;
 use futures::task::{Context, Poll};
 use log::*;
@@ -141,7 +141,6 @@ pub(crate) struct SocksClient {
     service_provider: Recipient,
     self_address: Recipient,
     started_proxy: bool,
-    lane_queue_lengths: LaneQueueLengths,
     shutdown_listener: ShutdownListener,
 }
 
@@ -159,7 +158,6 @@ impl Drop for SocksClient {
 
 impl SocksClient {
     /// Create a new SOCKClient
-    #[allow(clippy::too_many_arguments)]
     pub fn new(
         stream: TcpStream,
         authenticator: Authenticator,
@@ -167,7 +165,6 @@ impl SocksClient {
         service_provider: Recipient,
         controller_sender: ControllerSender,
         self_address: Recipient,
-        lane_queue_lengths: LaneQueueLengths,
         shutdown_listener: ShutdownListener,
     ) -> Self {
         let connection_id = Self::generate_random();
@@ -182,7 +179,6 @@ impl SocksClient {
             service_provider,
             self_address,
             started_proxy: false,
-            lane_queue_lengths,
             shutdown_listener,
         }
     }
@@ -202,7 +198,6 @@ impl SocksClient {
     pub async fn shutdown(&mut self) -> Result<(), SocksProxyError> {
         info!("client is shutting down its TCP stream");
         self.stream.shutdown().await?;
-        self.shutdown_listener.mark_as_success();
         Ok(())
     }
 
@@ -235,16 +230,8 @@ impl SocksClient {
         let req = Request::new_connect(self.connection_id, remote_address, self.self_address);
         let msg = Message::Request(req);
 
-        let input_message = InputMessage::new_fresh(
+        let input_message = InputMessage::new_fresh(self.service_provider, msg.into_bytes(), false);
-            self.service_provider,
+        self.input_sender.unbounded_send(input_message).unwrap();
-            msg.into_bytes(),
-            false,
-            TransmissionLane::ConnectionId(self.connection_id),
-        );
-        self.input_sender
-            .send(input_message)
-            .await
-            .expect("InputMessageReceiver has stopped receiving!");
     }
 
     async fn run_proxy(&mut self, conn_receiver: ConnectionReceiver, remote_proxy_target: String) {
@@ -267,14 +254,12 @@ impl SocksClient {
             conn_receiver,
             input_sender,
             connection_id,
-            Some(self.lane_queue_lengths.clone()),
             self.shutdown_listener.clone(),
         )
         .run(move |conn_id, read_data, socket_closed| {
             let provider_request = Request::new_send(conn_id, read_data, socket_closed);
             let provider_message = Message::Request(provider_request);
-            let lane = TransmissionLane::ConnectionId(conn_id);
+            InputMessage::new_fresh(recipient, provider_message.into_bytes(), false)
-            InputMessage::new_fresh(recipient, provider_message.into_bytes(), false, lane)
         })
         .await
         .into_inner();
@@ -319,7 +304,6 @@ impl SocksClient {
             SocksCommand::UdpAssociate => unimplemented!(), // not handled
         };
 
-        self.shutdown_listener.mark_as_success();
         Ok(())
     }
 

clients/socks5/src/socks/mixnet_responses.rs

@@ -1,5 +1,3 @@
-use std::time::Duration;
-
 use futures::channel::mpsc;
 use futures::StreamExt;
 use log::*;
@@ -20,16 +18,9 @@ pub(crate) struct MixnetResponseListener {
 
 impl Drop for MixnetResponseListener {
     fn drop(&mut self) {
-        if let Err(err) = self
+        self.buffer_requester
-            .buffer_requester
             .unbounded_send(ReceivedBufferMessage::ReceiverDisconnect)
-        {
+            .expect("the buffer request failed!")
-            if self.shutdown.is_shutdown_poll() {
-                log::debug!("The buffer request failed: {}", err);
-            } else {
-                log::error!("The buffer request failed: {}", err);
-            }
-        }
     }
 }
 
@@ -105,9 +96,7 @@ impl MixnetResponseListener {
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), self.shutdown.recv())
+        assert!(self.shutdown.is_shutdown_poll());
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("MixnetResponseListener: Exiting");
     }
 }

clients/socks5/src/socks/server.rs

@@ -1,17 +1,16 @@
-use crate::error::Socks5ClientError;
-
 use super::authentication::Authenticator;
 use super::client::SocksClient;
-use super::{mixnet_responses::MixnetResponseListener, types::ResponseCode};
+use super::{
-use client_connections::{ConnectionCommandSender, LaneQueueLengths};
+    mixnet_responses::MixnetResponseListener,
+    types::{ResponseCode, SocksProxyError},
+};
 use client_core::client::{
     inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender,
 };
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
-use proxy_helpers::connection_controller::{BroadcastActiveConnections, Controller};
+use proxy_helpers::connection_controller::Controller;
 use std::net::SocketAddr;
-use tap::TapFallible;
 use task::ShutdownListener;
 use tokio::net::TcpListener;
 
@@ -21,7 +20,6 @@ pub struct SphinxSocksServer {
     listening_address: SocketAddr,
     service_provider: Recipient,
     self_address: Recipient,
-    lane_queue_lengths: LaneQueueLengths,
     shutdown: ShutdownListener,
 }
 
@@ -32,7 +30,6 @@ impl SphinxSocksServer {
         authenticator: Authenticator,
         service_provider: Recipient,
         self_address: Recipient,
-        lane_queue_lengths: LaneQueueLengths,
         shutdown: ShutdownListener,
     ) -> Self {
         // hardcode ip as we (presumably) ONLY want to listen locally. If we change it, we can
@@ -44,7 +41,6 @@ impl SphinxSocksServer {
             listening_address: format!("{}:{}", ip, port).parse().unwrap(),
             service_provider,
             self_address,
-            lane_queue_lengths,
             shutdown,
         }
     }
@@ -55,19 +51,13 @@ impl SphinxSocksServer {
         &mut self,
         input_sender: InputMessageSender,
         buffer_requester: ReceivedBufferRequestSender,
-        client_connection_tx: ConnectionCommandSender,
+    ) -> Result<(), SocksProxyError> {
-    ) -> Result<(), Socks5ClientError> {
+        let listener = TcpListener::bind(self.listening_address).await.unwrap();
-        let listener = TcpListener::bind(self.listening_address)
-            .await
-            .tap_err(|err| log::error!("Failed to bind to address: {err}"))?;
         info!("Serving Connections...");
 
         // controller for managing all active connections
-        let (mut active_streams_controller, controller_sender) = Controller::new(
+        let (mut active_streams_controller, controller_sender) =
-            client_connection_tx,
+            Controller::new(self.shutdown.clone());
-            BroadcastActiveConnections::Off,
-            self.shutdown.clone(),
-        );
         tokio::spawn(async move {
             active_streams_controller.run().await;
         });
@@ -93,7 +83,6 @@ impl SphinxSocksServer {
                         self.service_provider,
                         controller_sender.clone(),
                         self.self_address,
-                        self.lane_queue_lengths.clone(),
                         self.shutdown.clone(),
                     );
 

clients/webassembly/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "nym-client-wasm"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jedrzej Stuczynski <andrew@nymtech.net>"]
-version = "1.1.0"
+version = "1.0.1"
 edition = "2021"
 keywords = ["nym", "sphinx", "wasm", "webassembly", "privacy", "client"]
 license = "Apache-2.0"
@@ -19,18 +19,16 @@ coconut = ["coconut-interface", "credentials", "gateway-client/coconut"]
 
 [dependencies]
 futures = "0.3"
-js-sys = "0.3"
-rand = { version = "0.7.3", features = ["wasm-bindgen"] }
 serde = { version = "1.0", features = ["derive"] }
 serde-wasm-bindgen = "0.4"
-tokio = { version = "1.21.2", features = ["sync"] }
-url = "2.2"
 wasm-bindgen = { version = "=0.2.83", features = ["serde-serialize"]  }
 wasm-bindgen-futures = "0.4"
+js-sys = "0.3"
+rand = { version = "0.7.3", features = ["wasm-bindgen"] }
+url = "2.2"
 
 # internal
 client-core = { path = "../client-core", default-features = false, features = ["wasm"] }
-client-connections = { path = "../../common/client-connections" }
 coconut-interface = { path = "../../common/coconut-interface", optional = true }
 credentials = { path = "../../common/credentials", optional = true }
 crypto = { path = "../../common/crypto" }
@@ -61,4 +59,4 @@ wasm-opt = true
 
 [profile.release]
 lto = true
-opt-level = 'z'
+opt-level = 'z'

clients/webassembly/Makefile

@@ -1,5 +0,0 @@
-clippy:
-	cargo clippy --all-features --target wasm32-unknown-unknown -- -D warnings
-
-test:
-	wasm-pack test --node

clients/webassembly/src/binary_message_helper.rs

@@ -1,226 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use serde::{Deserialize, Serialize};
-use wasm_bindgen::prelude::*;
-use wasm_bindgen::JsCast;
-
-#[wasm_bindgen(typescript_custom_section)]
-const TS_DEFS: &'static str = r#"
-export interface BinaryMessage {
-    kind: number,
-    payload: Uint8Array;
-    headers: string,
-}
-
-export interface StringMessage {
-    kind: number,
-    payload: string;
-}
-"#;
-
-#[wasm_bindgen]
-extern "C" {
-    #[wasm_bindgen(typescript_type = "BinaryMessage")]
-    pub type IBinaryMessage;
-    #[wasm_bindgen(typescript_type = "StringMessage")]
-    pub type IStringMessage;
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct BinaryMessage {
-    pub kind: u8,
-    pub payload: Vec<u8>,
-    pub headers: String,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct StringMessage {
-    pub kind: u8,
-    pub payload: String,
-}
-
-/// Create a new binary message with a user-specified `kind`.
-#[wasm_bindgen]
-pub fn create_binary_message(kind: u8, payload: Vec<u8>, headers: Option<String>) -> Vec<u8> {
-    create_binary_message_with_headers(kind, payload, headers)
-}
-
-/// Create a new message with a UTF-8 encoded string `payload` and a user-specified `kind`.
-#[wasm_bindgen]
-pub fn create_binary_message_from_string(
-    kind: u8,
-    payload: String,
-    headers: Option<String>,
-) -> Vec<u8> {
-    create_binary_message_with_headers(kind, payload.as_bytes().to_vec(), headers)
-}
-
-/// Create a new binary message with a user-specified `kind`, and `headers` as a string.
-#[wasm_bindgen]
-pub fn create_binary_message_with_headers(
-    kind: u8,
-    payload: Vec<u8>,
-    headers: Option<String>,
-) -> Vec<u8> {
-    let headers = headers.unwrap_or_else(|| "".to_string());
-    let headers = headers.as_bytes().to_vec();
-    let size = (headers.len() as u64).to_be_bytes().to_vec();
-    vec![vec![kind], size, headers, payload].concat()
-}
-
-/// Parse the `kind` and byte array `payload` from a byte array
-#[wasm_bindgen]
-pub async fn parse_binary_message(message: Vec<u8>) -> Result<IBinaryMessage, JsError> {
-    if message.len() < 2 {
-        return Err(JsError::new(
-            "Could not parse message, as less than 2 bytes long",
-        ));
-    }
-
-    let (kind, _headers, payload) = parse_binary_payload(&message);
-
-    Ok(serde_wasm_bindgen::to_value(&BinaryMessage {
-        kind,
-        payload: payload.to_vec(),
-        headers: "".to_string(),
-    })
-    .unwrap()
-    .unchecked_into::<IBinaryMessage>())
-}
-
-/// Parse the `kind` and byte array `payload` from a byte array with headers
-#[wasm_bindgen]
-pub async fn parse_binary_message_with_headers(
-    message: Vec<u8>,
-) -> Result<IBinaryMessage, JsError> {
-    if message.len() < 2 {
-        return Err(JsError::new(
-            "Could not parse message, as less than 2 bytes long",
-        ));
-    }
-
-    let (kind, headers, payload) = parse_binary_payload(&message);
-
-    Ok(serde_wasm_bindgen::to_value(&BinaryMessage {
-        kind,
-        payload: payload.to_vec(),
-        headers,
-    })
-    .unwrap()
-    .unchecked_into::<IBinaryMessage>())
-}
-
-/// Parse the `kind` and UTF-8 string `payload` from a byte array with headers
-#[wasm_bindgen]
-pub async fn parse_string_message_with_headers(
-    message: Vec<u8>,
-) -> Result<IStringMessage, JsError> {
-    if message.len() < 2 {
-        return Err(JsError::new(
-            "Could not parse message, as less than 2 bytes long",
-        ));
-    }
-
-    let (kind, _headers, payload) = parse_binary_payload(&message);
-    let payload = String::from_utf8_lossy(payload).into_owned();
-
-    Ok(
-        serde_wasm_bindgen::to_value(&StringMessage { kind, payload })
-            .unwrap()
-            .unchecked_into::<IStringMessage>(),
-    )
-}
-pub(crate) fn parse_binary_payload(message: &[u8]) -> (u8, String, &[u8]) {
-    // 1st byte is the kind
-    let kind = message[0];
-
-    // then the size as u64 big endian
-    let mut size = [0u8; 8];
-    size.clone_from_slice(&message[1..9]);
-    let size = u64::from_be_bytes(size) as usize;
-
-    // then the headers
-    let headers = String::from_utf8_lossy(&message[9..9 + size]).into_owned();
-
-    // finally the payload
-    let payload = &message[9 + size..];
-
-    (kind, headers, payload)
-}
-
-/// Parse the `kind` and UTF-8 string `payload` from a byte array
-#[wasm_bindgen]
-pub async fn parse_string_message(message: Vec<u8>) -> Result<IStringMessage, JsError> {
-    if message.len() < 2 {
-        return Err(JsError::new(
-            "Could not parse message, as less than 2 bytes long",
-        ));
-    }
-
-    let kind = message[0];
-    let payload = String::from_utf8_lossy(&message[1..]).into_owned();
-
-    Ok(
-        serde_wasm_bindgen::to_value(&StringMessage { kind, payload })
-            .unwrap()
-            .unchecked_into::<IStringMessage>(),
-    )
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{create_binary_message_with_headers, parse_binary_payload};
-    use wasm_bindgen_test::*;
-
-    #[wasm_bindgen_test]
-    fn test_binary_with_headers() {
-        let message_as_bytes = create_binary_message_with_headers(
-            42u8,
-            vec![0u8, 1u8, 2u8],
-            Some("test headers".to_string()),
-        );
-
-        // calculate header size
-        let headers = "test headers".as_bytes().to_vec();
-        let size = headers.len();
-
-        // the expected size
-        let expected_size = 12;
-        assert_eq!(size, expected_size);
-
-        assert_eq!(message_as_bytes[0], 42u8);
-        assert_eq!(message_as_bytes[1..9], 12u64.to_be_bytes());
-        assert_eq!(
-            message_as_bytes[9 + expected_size..9 + expected_size + 3],
-            vec![0u8, 1u8, 2u8]
-        );
-
-        let res = parse_binary_payload(&message_as_bytes);
-
-        assert_eq!(res.0, 42u8);
-        assert_eq!(res.1, "test headers".to_string());
-        assert_eq!(res.2, vec![0u8, 1u8, 2u8]);
-    }
-
-    #[wasm_bindgen_test]
-    fn test_binary_with_empty_headers() {
-        let message_as_bytes =
-            create_binary_message_with_headers(42u8, vec![0u8, 1u8, 2u8], Some("".to_string()));
-
-        let expected_size = 0;
-
-        assert_eq!(message_as_bytes[0], 42u8);
-        assert_eq!(message_as_bytes[1..9], 0u64.to_be_bytes());
-        assert_eq!(
-            message_as_bytes[9 + expected_size..9 + expected_size + 3],
-            vec![0u8, 1u8, 2u8]
-        );
-
-        let res = parse_binary_payload(&message_as_bytes);
-
-        assert_eq!(res.0, 42u8);
-        assert_eq!(res.1, "".to_string());
-        assert_eq!(res.2, vec![0u8, 1u8, 2u8]);
-    }
-}

clients/webassembly/src/client/mod.rs

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use self::config::Config;
-use client_connections::{ConnectionCommandReceiver, LaneQueueLengths, TransmissionLane};
 use client_core::client::{
     cover_traffic_stream::LoopCoverTrafficStream,
     inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender},
@@ -26,7 +25,7 @@ use nymsphinx::addressing::clients::Recipient;
 use rand::rngs::OsRng;
 use wasm_bindgen::prelude::*;
 use wasm_bindgen_futures::spawn_local;
-use wasm_utils::console_log;
+use wasm_utils::{console_log, console_warn};
 
 pub mod config;
 
@@ -45,7 +44,6 @@ pub struct NymClient {
 
     // callbacks
     on_message: Option<js_sys::Function>,
-    on_binary_message: Option<js_sys::Function>,
     on_gateway_connect: Option<js_sys::Function>,
 }
 
@@ -57,7 +55,6 @@ impl NymClient {
             config,
             key_manager: Self::setup_key_manager(),
             on_message: None,
-            on_binary_message: None,
             on_gateway_connect: None,
             input_tx: None,
         }
@@ -75,11 +72,8 @@ impl NymClient {
         self.on_message = Some(on_message);
     }
 
-    pub fn set_on_binary_message(&mut self, on_binary_message: js_sys::Function) {
-        self.on_binary_message = Some(on_binary_message);
-    }
-
     pub fn set_on_gateway_connect(&mut self, on_connect: js_sys::Function) {
+        console_log!("setting on connect...");
         self.on_gateway_connect = Some(on_connect)
     }
 
@@ -128,8 +122,6 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
-        client_connection_rx: ConnectionCommandReceiver,
-        lane_queue_lengths: LaneQueueLengths,
     ) {
         let mut controller_config = real_messages_control::Config::new(
             self.key_manager.ack_key(),
@@ -154,8 +146,6 @@ impl NymClient {
             input_receiver,
             mix_sender,
             topology_accessor,
-            lane_queue_lengths,
-            client_connection_rx,
         )
         .start();
     }
@@ -276,7 +266,6 @@ impl NymClient {
         received_buffer_request_sender: ReceivedBufferRequestSender,
     ) {
         let on_message = self.on_message.take();
-        let on_binary_message = self.on_binary_message.take();
 
         spawn_local(async move {
             let (reconstructed_sender, mut reconstructed_receiver) = mpsc::unbounded();
@@ -291,14 +280,8 @@ impl NymClient {
             let this = JsValue::null();
 
             while let Some(reconstructed) = reconstructed_receiver.next().await {
-                for msg in reconstructed {
+                if let Some(ref callback) = on_message {
-                    if let Some(ref callback_binary) = on_binary_message {
+                    for msg in reconstructed {
-                        let arg1 = serde_wasm_bindgen::to_value(&msg.message).unwrap();
-                        callback_binary
-                            .call1(&this, &arg1)
-                            .expect("on binary message failed!");
-                    }
-                    if let Some(ref callback) = on_message {
                         if msg.reply_surb.is_some() {
                             console_log!("the received message contained a reply-surb that we do not know how to handle (yet)")
                         }
@@ -306,6 +289,9 @@ impl NymClient {
                         let arg1 = serde_wasm_bindgen::to_value(&stringified).unwrap();
                         callback.call1(&this, &arg1).expect("on message failed!");
                     }
+                } else {
+                    console_warn!("no on_message callback was specified. the received message content is getting dropped");
+                    console_log!("the raw messages: {:?}", reconstructed)
                 }
             }
         });
@@ -326,16 +312,12 @@ impl NymClient {
         let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
 
         // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
+        let (input_sender, input_receiver) = mpsc::unbounded::<InputMessage>();
 
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
         let shared_topology_accessor = TopologyAccessor::new();
 
-        // Channel that the real traffix controller can listed to for closing connections.
-        // Currently unused in the wasm client.
-        let (_client_connection_tx, client_connection_rx) = mpsc::unbounded();
-
         // the components are started in very specific order. Unless you know what you are doing,
         // do not change that.
         self.start_topology_refresher(shared_topology_accessor.clone())
@@ -355,17 +337,11 @@ impl NymClient {
         // The MixTrafficController then sends the actual traffic
         let sphinx_message_sender = Self::start_mix_traffic_controller(gateway_client);
 
-        // Shared queue length data. Published by the `OutQueueController` in the client, and used
-        // primarily to throttle incoming connections
-        let shared_lane_queue_lengths = LaneQueueLengths::new();
-
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
-            client_connection_rx,
-            shared_lane_queue_lengths,
         );
 
         if !self.config.debug.disable_loop_cover_traffic_stream {
@@ -391,16 +367,14 @@ impl NymClient {
         console_log!("Sending {} bytes to {}", message.len(), recipient);
 
         let recipient = Recipient::try_from_base58_string(recipient).unwrap();
-        let lane = TransmissionLane::General;
 
-        let input_msg = InputMessage::new_fresh(recipient, message, false, lane);
+        let input_msg = InputMessage::new_fresh(recipient, message, false);
 
         self.input_tx
             .as_ref()
             .expect("start method was not called before!")
-            .send(input_msg)
+            .unbounded_send(input_msg)
-            .await
+            .unwrap();
-            .expect("InputMessageReceiver has stopped receiving!");
 
         self
     }

clients/webassembly/src/gateway_selector.rs

@@ -6,7 +6,7 @@ use wasm_bindgen::prelude::*;
 
 #[wasm_bindgen]
 pub async fn get_gateway(api_server: String, preferred: Option<String>) -> GatewayEndpoint {
-    let validator_client = validator_client::client::ApiClient::new(api_server.parse().unwrap());
+    let validator_client = validator_client::ApiClient::new(api_server.parse().unwrap());
 
     let gateways = match validator_client.get_cached_gateways().await {
         Err(err) => panic!("failed to obtain list of all gateways - {}", err),

clients/webassembly/src/lib.rs

@@ -3,8 +3,6 @@
 
 use wasm_bindgen::prelude::*;
 
-#[cfg(target_arch = "wasm32")]
-pub mod binary_message_helper;
 #[cfg(target_arch = "wasm32")]
 mod client;
 #[cfg(target_arch = "wasm32")]

common/client-connections/Cargo.toml

@@ -1,10 +0,0 @@
-[package]
-name = "client-connections"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-futures = "0.3"
-log = "0.4.17"

common/client-connections/src/lib.rs

@@ -1,111 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use std::collections::HashMap;
-
-use futures::channel::mpsc;
-
-pub type ConnectionId = u64;
-
-#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)]
-pub enum TransmissionLane {
-    General,
-    Reply,
-    Retransmission,
-    Control,
-    ConnectionId(ConnectionId),
-}
-
-/// Used by the connection controller to report current state for client connections.
-pub type ConnectionCommandSender = mpsc::UnboundedSender<ConnectionCommand>;
-pub type ConnectionCommandReceiver = mpsc::UnboundedReceiver<ConnectionCommand>;
-
-pub enum ConnectionCommand {
-    // Announce that at a connection was closed. E.g the `OutQueueControl` uses this to discard
-    // transmission lanes.
-    Close(ConnectionId),
-
-    // In the network requester for example, we usually want to broadcast active connections
-    // regularly, so we know what connections we need to request lane queue lengths for from the
-    // client.
-    // In the socks5-client, this is not needed since have direct access to the lane queue lengths.
-    ActiveConnections(Vec<ConnectionId>),
-}
-
-// The `OutQueueControl` publishes the backlog per lane, primarily so that upstream can slow down
-// if needed.
-#[derive(Clone, Debug)]
-pub struct LaneQueueLengths(std::sync::Arc<std::sync::Mutex<LaneQueueLengthsInner>>);
-
-impl LaneQueueLengths {
-    pub fn new() -> Self {
-        LaneQueueLengths(std::sync::Arc::new(std::sync::Mutex::new(
-            LaneQueueLengthsInner {
-                map: HashMap::new(),
-            },
-        )))
-    }
-
-    pub fn set(&mut self, lane: &TransmissionLane, lane_length: Option<usize>) {
-        match self.0.lock() {
-            Ok(mut inner) => {
-                if let Some(length) = lane_length {
-                    inner
-                        .map
-                        .entry(*lane)
-                        .and_modify(|e| *e = length)
-                        .or_insert(length);
-                } else {
-                    inner.map.remove(lane);
-                }
-            }
-            Err(err) => log::warn!("Failed to set lane queue length: {err}"),
-        }
-    }
-
-    pub fn get(&self, lane: &TransmissionLane) -> Option<usize> {
-        match self.0.lock() {
-            Ok(inner) => inner.get(lane),
-            Err(err) => {
-                log::warn!("Failed to get lane queue length: {err}");
-                None
-            }
-        }
-    }
-}
-
-impl Default for LaneQueueLengths {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl std::ops::Deref for LaneQueueLengths {
-    type Target = std::sync::Arc<std::sync::Mutex<LaneQueueLengthsInner>>;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-#[derive(Debug)]
-pub struct LaneQueueLengthsInner {
-    pub map: HashMap<TransmissionLane, usize>,
-}
-
-impl LaneQueueLengthsInner {
-    pub fn get(&self, lane: &TransmissionLane) -> Option<usize> {
-        self.map.get(lane).copied()
-    }
-
-    pub fn values(&self) -> impl Iterator<Item = &usize> {
-        self.map.values()
-    }
-
-    pub fn modify<F>(&mut self, lane: &TransmissionLane, f: F)
-    where
-        F: FnOnce(&mut usize),
-    {
-        self.map.entry(*lane).and_modify(f);
-    }
-}

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

@@ -17,8 +17,6 @@ use credential_storage::error::StorageError;
 #[cfg(feature = "coconut")]
 use std::str::FromStr;
 #[cfg(feature = "coconut")]
-use validator_client::client::CoconutApiClient;
-#[cfg(feature = "coconut")]
 use {
     coconut_interface::Base58,
     credentials::coconut::{
@@ -31,7 +29,7 @@ pub struct BandwidthController<St: Storage> {
     #[allow(dead_code)]
     storage: St,
     #[cfg(feature = "coconut")]
-    coconut_api_clients: Vec<CoconutApiClient>,
+    validator_endpoints: Vec<url::Url>,
 }
 
 impl<St> BandwidthController<St>
@@ -39,10 +37,10 @@ where
     St: Storage + Clone + 'static,
 {
     #[cfg(feature = "coconut")]
-    pub fn new(storage: St, coconut_api_clients: Vec<CoconutApiClient>) -> Self {
+    pub fn new(storage: St, validator_endpoints: Vec<url::Url>) -> Self {
         BandwidthController {
             storage,
-            coconut_api_clients,
+            validator_endpoints,
         }
     }
 
@@ -55,7 +53,7 @@ where
     pub async fn prepare_coconut_credential(
         &self,
     ) -> Result<coconut_interface::Credential, GatewayClientError> {
-        let verification_key = obtain_aggregate_verification_key(&self.coconut_api_clients).await?;
+        let verification_key = obtain_aggregate_verification_key(&self.validator_endpoints).await?;
         let bandwidth_credential = self.storage.get_next_coconut_credential().await?;
         let voucher_value = u64::from_str(&bandwidth_credential.voucher_value)
             .map_err(|_| StorageError::InconsistentData)?;

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

@@ -2,19 +2,25 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::bandwidth::BandwidthController;
+use crate::cleanup_socket_message;
 use crate::error::GatewayClientError;
 use crate::packet_router::PacketRouter;
 pub use crate::packet_router::{
     AcknowledgementReceiver, AcknowledgementSender, MixnetMessageReceiver, MixnetMessageSender,
 };
 use crate::socket_state::{PartiallyDelegated, SocketState};
-use crate::{cleanup_socket_message, try_decrypt_binary_message};
+#[cfg(target_arch = "wasm32")]
+use crate::wasm_storage::PersistentStorage;
+#[cfg(feature = "coconut")]
+use coconut_interface::Credential;
+#[cfg(not(target_arch = "wasm32"))]
+use credential_storage::PersistentStorage;
 use crypto::asymmetric::identity;
 use futures::{FutureExt, SinkExt, StreamExt};
 use gateway_requests::authentication::encrypted_address::EncryptedAddressBytes;
 use gateway_requests::iv::IV;
 use gateway_requests::registration::handshake::{client_handshake, SharedKeys};
-use gateway_requests::{BinaryRequest, ClientControlRequest, ServerResponse, PROTOCOL_VERSION};
+use gateway_requests::{BinaryRequest, ClientControlRequest, ServerResponse};
 use log::*;
 use network_defaults::{REMAINING_BANDWIDTH_THRESHOLD, TOKENS_TO_BURN};
 use nymsphinx::forwarding::packet::MixPacket;
@@ -22,20 +28,13 @@ use rand::rngs::OsRng;
 use std::convert::TryFrom;
 use std::sync::Arc;
 use std::time::Duration;
-use tungstenite::protocol::Message;
-
-#[cfg(feature = "coconut")]
-use coconut_interface::Credential;
-
-#[cfg(not(target_arch = "wasm32"))]
-use credential_storage::PersistentStorage;
 #[cfg(not(target_arch = "wasm32"))]
 use task::ShutdownListener;
+use tungstenite::protocol::Message;
+
 #[cfg(not(target_arch = "wasm32"))]
 use tokio_tungstenite::connect_async;
 
-#[cfg(target_arch = "wasm32")]
-use crate::wasm_storage::PersistentStorage;
 #[cfg(target_arch = "wasm32")]
 use wasm_timer;
 #[cfg(target_arch = "wasm32")]
@@ -307,8 +306,6 @@ impl GatewayClient {
             let m_shutdown = self.shutdown.clone();
             async {
                 if let Some(mut s) = m_shutdown {
-                    // TODO: fix this by marking as success _after_ the select
-                    s.mark_as_success();
                     s.recv().await
                 } else {
                     std::future::pending::<()>().await
@@ -339,15 +336,7 @@ impl GatewayClient {
                     };
                     match ws_msg {
                         Message::Binary(bin_msg) => {
-                            // if we have established the shared key already, attempt to use it for decryption
+                            if let Err(err) = self.packet_router.route_received(vec![bin_msg]) {
-                            // otherwise there's not much we can do apart from just routing what we have on hand
-                            if let Some(shared_keys) = &self.shared_key {
-                                if let Some(plaintext) = try_decrypt_binary_message(bin_msg, shared_keys) {
-                                    if let Err(err) = self.packet_router.route_received(vec![plaintext]) {
-                                        log::warn!("Route received failed: {:?}", err);
-                                    }
-                                }
-                            } else if let Err(err) = self.packet_router.route_received(vec![bin_msg]) {
                                 log::warn!("Route received failed: {:?}", err);
                             }
                         }
@@ -447,33 +436,6 @@ impl GatewayClient {
         }
     }
 
-    fn check_gateway_protocol(
-        &self,
-        gateway_protocol: Option<u8>,
-    ) -> Result<(), GatewayClientError> {
-        // right now there are no failure cases here, but this might change in the future
-        match gateway_protocol {
-            None => {
-                warn!("the gateway we're connected to has not specified its protocol version. It's probably running version < 1.1.X, but that's still fine for now. It will become a hard error in 1.2.0");
-                // note: in 1.2.0 we will have to return a hard error here
-                Ok(())
-            }
-            Some(v) if v != PROTOCOL_VERSION => {
-                let err = GatewayClientError::IncompatibleProtocol {
-                    gateway: Some(v),
-                    current: PROTOCOL_VERSION,
-                };
-                error!("{err}");
-                Err(err)
-            }
-
-            Some(_) => {
-                info!("the gateway is using exactly the same protocol version as we are. We're good to continue!");
-                Ok(())
-            }
-        }
-    }
-
     async fn register(&mut self) -> Result<(), GatewayClientError> {
         if !self.connection.is_established() {
             return Err(GatewayClientError::ConnectionNotEstablished);
@@ -496,20 +458,11 @@ impl GatewayClient {
             .map_err(GatewayClientError::RegistrationFailure),
             _ => unreachable!(),
         }?;
-        let (authentication_status, gateway_protocol) = match self.read_control_response().await? {
+        self.authenticated = match self.read_control_response().await? {
-            ServerResponse::Register {
+            ServerResponse::Register { status } => Ok(status),
-                protocol_version,
+            ServerResponse::Error { message } => Err(GatewayClientError::GatewayError(message)),
-                status,
+            _ => Err(GatewayClientError::UnexpectedResponse),
-            } => (status, protocol_version),
+        }?;
-            ServerResponse::Error { message } => {
-                return Err(GatewayClientError::GatewayError(message))
-            }
-            _ => return Err(GatewayClientError::UnexpectedResponse),
-        };
-
-        self.check_gateway_protocol(gateway_protocol)?;
-        self.authenticated = authentication_status;
-
         if self.authenticated {
             self.shared_key = Some(Arc::new(shared_key));
         }
@@ -548,11 +501,9 @@ impl GatewayClient {
 
         match self.send_websocket_message(msg).await? {
             ServerResponse::Authenticate {
-                protocol_version,
                 status,
                 bandwidth_remaining,
             } => {
-                self.check_gateway_protocol(protocol_version)?;
                 self.authenticated = status;
                 self.bandwidth_remaining = bandwidth_remaining;
                 Ok(())

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

@@ -85,9 +85,6 @@ pub enum GatewayClientError {
 
     #[error("Failed to send mixnet message")]
     MixnetMsgSenderFailedToSend,
-
-    #[error("Attempted to negotiate connection with gateway using incompatible protocol version. Ours is {current} and the gateway reports {gateway:?}")]
-    IncompatibleProtocol { gateway: Option<u8>, current: u8 },
 }
 
 impl GatewayClientError {

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

@@ -3,9 +3,6 @@
 
 use crate::error::GatewayClientError;
 pub use client::GatewayClient;
-use gateway_requests::registration::handshake::SharedKeys;
-use gateway_requests::BinaryResponse;
-use log::warn;
 pub use packet_router::{
     AcknowledgementReceiver, AcknowledgementSender, MixnetMessageReceiver, MixnetMessageSender,
 };
@@ -40,21 +37,3 @@ pub(crate) fn cleanup_socket_messages(
         None => Err(GatewayClientError::ConnectionAbruptlyClosed),
     }
 }
-
-pub(crate) fn try_decrypt_binary_message(
-    bin_msg: Vec<u8>,
-    shared_keys: &SharedKeys,
-) -> Option<Vec<u8>> {
-    match BinaryResponse::try_from_encrypted_tagged_bytes(bin_msg, shared_keys) {
-        Ok(bin_response) => match bin_response {
-            BinaryResponse::PushedMixMessage(plaintext) => Some(plaintext),
-        },
-        Err(err) => {
-            warn!(
-                "message received from the gateway was malformed! - {:?}",
-                err
-            );
-            None
-        }
-    }
-}

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

@@ -1,13 +1,14 @@
 // Copyright 2021-2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::cleanup_socket_messages;
 use crate::error::GatewayClientError;
 use crate::packet_router::PacketRouter;
-use crate::{cleanup_socket_messages, try_decrypt_binary_message};
 use futures::channel::oneshot;
 use futures::stream::{SplitSink, SplitStream};
 use futures::{SinkExt, StreamExt};
 use gateway_requests::registration::handshake::SharedKeys;
+use gateway_requests::BinaryResponse;
 use log::*;
 use std::sync::Arc;
 #[cfg(not(target_arch = "wasm32"))]
@@ -49,9 +50,21 @@ impl PartiallyDelegated {
             match ws_msg {
                 Message::Binary(bin_msg) => {
                     // this function decrypts the request and checks the MAC
-                    if let Some(plaintext) = try_decrypt_binary_message(bin_msg, shared_key) {
+                    let plaintext = match BinaryResponse::try_from_encrypted_tagged_bytes(
-                        plaintexts.push(plaintext)
+                        bin_msg, shared_key,
-                    }
+                    ) {
+                        Ok(bin_response) => match bin_response {
+                            BinaryResponse::PushedMixMessage(plaintext) => plaintext,
+                        },
+                        Err(err) => {
+                            warn!(
+                                "message received from the gateway was malformed! - {:?}",
+                                err
+                            );
+                            continue;
+                        }
+                    };
+                    plaintexts.push(plaintext)
                 }
                 // I think that in the future we should perhaps have some sequence number system, i.e.
                 // so each request/response pair can be easily identified, so that if messages are

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

@@ -10,11 +10,10 @@ rust-version = "1.56"
 [dependencies]
 base64 = "0.13"
 colored = "2.0"
-
+cw3 = "0.13.1"
-coconut-dkg-common = { path = "../../cosmwasm-smart-contracts/coconut-dkg" }
-contracts-common = { path = "../../cosmwasm-smart-contracts/contracts-common" }
 mixnet-contract-common = { path= "../../cosmwasm-smart-contracts/mixnet-contract" }
 vesting-contract-common = { path= "../../cosmwasm-smart-contracts/vesting-contract" }
+contracts-common = { path = "../../cosmwasm-smart-contracts/contracts-common" }
 coconut-bandwidth-contract-common = { path= "../../cosmwasm-smart-contracts/coconut-bandwidth-contract" }
 multisig-contract-common = { path = "../../cosmwasm-smart-contracts/multisig-contract" }
 vesting-contract = { path = "../../../contracts/vesting" }
@@ -38,7 +37,6 @@ async-trait = { version = "0.1.51", optional = true }
 bip39 = { version = "1", features = ["rand"], optional = true }
 config = { path = "../../config", optional = true }
 cosmrs = { git = "https://github.com/neacsu/cosmos-rust", branch = "neacsu/feegrant_support", features = ["rpc", "bip32", "cosmwasm"], optional = true}
-cw3 = { version = "0.13.4", optional = true }
 prost = { version = "0.10", default-features = false, optional = true }
 flate2 = { version = "1.0.20", optional = true }
 sha2 = { version = "0.9.5", optional = true }
@@ -55,7 +53,6 @@ nymd-client = [
     "bip39",
     "config",
     "cosmrs",
-    "cw3",
     "prost",
     "flate2",
     "sha2",

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

@@ -2,21 +2,13 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{validator_api, ValidatorClientError};
-use coconut_dkg_common::types::NodeIndex;
-#[cfg(feature = "nymd-client")]
-use coconut_dkg_common::{
-    dealer::ContractDealing, types::DealerDetails, verification_key::ContractVKShare,
-};
-#[cfg(feature = "nymd-client")]
-use coconut_interface::Base58;
-use coconut_interface::VerificationKey;
 use mixnet_contract_common::mixnode::MixNodeDetails;
 use mixnet_contract_common::MixId;
 use mixnet_contract_common::{GatewayBond, IdentityKeyRef};
-#[cfg(feature = "nymd-client")]
+use url::Url;
-use std::str::FromStr;
 use validator_api_requests::coconut::{
-    BlindSignRequestBody, BlindedSignatureResponse, VerifyCredentialBody, VerifyCredentialResponse,
+    BlindSignRequestBody, BlindedSignatureResponse, CosmosAddressResponse, VerificationKeyResponse,
+    VerifyCredentialBody, VerifyCredentialResponse,
 };
 use validator_api_requests::models::{
     GatewayCoreStatusResponse, MixnodeCoreStatusResponse, MixnodeStatusResponse,
@@ -24,12 +16,10 @@ use validator_api_requests::models::{
 };
 
 #[cfg(feature = "nymd-client")]
-use crate::nymd::traits::{DkgQueryClient, MixnetQueryClient, MultisigQueryClient};
+use crate::nymd::traits::MixnetQueryClient;
 #[cfg(feature = "nymd-client")]
 use crate::nymd::{self, CosmWasmClient, NymdClient, QueryNymdClient, SigningNymdClient};
 #[cfg(feature = "nymd-client")]
-use cw3::ProposalResponse;
-#[cfg(feature = "nymd-client")]
 use mixnet_contract_common::{
     mixnode::MixNodeBond,
     pending_events::{PendingEpochEvent, PendingIntervalEvent},
@@ -37,7 +27,6 @@ use mixnet_contract_common::{
 };
 #[cfg(feature = "nymd-client")]
 use network_defaults::NymNetworkDetails;
-use url::Url;
 #[cfg(feature = "nymd-client")]
 use validator_api_requests::models::MixNodeBondAnnotated;
 
@@ -54,9 +43,6 @@ pub struct Config {
     gateway_page_limit: Option<u32>,
     mixnode_delegations_page_limit: Option<u32>,
     rewarded_set_page_limit: Option<u32>,
-    dealers_page_limit: Option<u32>,
-    verification_key_page_limit: Option<u32>,
-    proposals_page_limit: Option<u32>,
 }
 
 #[cfg(feature = "nymd-client")]
@@ -86,9 +72,6 @@ impl Config {
             gateway_page_limit: None,
             mixnode_delegations_page_limit: None,
             rewarded_set_page_limit: None,
-            dealers_page_limit: None,
-            verification_key_page_limit: None,
-            proposals_page_limit: None,
         })
     }
 
@@ -136,9 +119,6 @@ pub struct Client<C> {
     gateway_page_limit: Option<u32>,
     mixnode_delegations_page_limit: Option<u32>,
     rewarded_set_page_limit: Option<u32>,
-    dealers_page_limit: Option<u32>,
-    verification_key_page_limit: Option<u32>,
-    proposals_page_limit: Option<u32>,
 
     // ideally they would have been read-only, but unfortunately rust doesn't have such features
     pub validator_api: validator_api::Client,
@@ -165,9 +145,6 @@ impl Client<SigningNymdClient> {
             gateway_page_limit: config.gateway_page_limit,
             mixnode_delegations_page_limit: config.mixnode_delegations_page_limit,
             rewarded_set_page_limit: config.rewarded_set_page_limit,
-            dealers_page_limit: config.dealers_page_limit,
-            verification_key_page_limit: config.verification_key_page_limit,
-            proposals_page_limit: config.proposals_page_limit,
             validator_api: validator_api_client,
             nymd: nymd_client,
         })
@@ -201,9 +178,6 @@ impl Client<QueryNymdClient> {
             gateway_page_limit: config.gateway_page_limit,
             mixnode_delegations_page_limit: config.mixnode_delegations_page_limit,
             rewarded_set_page_limit: config.rewarded_set_page_limit,
-            dealers_page_limit: config.dealers_page_limit,
-            verification_key_page_limit: config.verification_key_page_limit,
-            proposals_page_limit: config.proposals_page_limit,
             validator_api: validator_api_client,
             nymd: nymd_client,
         })
@@ -545,135 +519,6 @@ impl<C> Client<C> {
 
         Ok(events)
     }
-
-    pub async fn get_all_nymd_current_dealers(
-        &self,
-    ) -> Result<Vec<DealerDetails>, ValidatorClientError>
-    where
-        C: CosmWasmClient + Sync + Send,
-    {
-        let mut dealers = Vec::new();
-        let mut start_after = None;
-        loop {
-            let mut paged_response = self
-                .nymd
-                .get_current_dealers_paged(start_after.take(), self.dealers_page_limit)
-                .await?;
-            dealers.append(&mut paged_response.dealers);
-
-            if let Some(start_after_res) = paged_response.start_next_after {
-                start_after = Some(start_after_res.into_string())
-            } else {
-                break;
-            }
-        }
-
-        Ok(dealers)
-    }
-
-    pub async fn get_all_nymd_past_dealers(
-        &self,
-    ) -> Result<Vec<DealerDetails>, ValidatorClientError>
-    where
-        C: CosmWasmClient + Sync + Send,
-    {
-        let mut dealers = Vec::new();
-        let mut start_after = None;
-        loop {
-            let mut paged_response = self
-                .nymd
-                .get_past_dealers_paged(start_after.take(), self.dealers_page_limit)
-                .await?;
-            dealers.append(&mut paged_response.dealers);
-
-            if let Some(start_after_res) = paged_response.start_next_after {
-                start_after = Some(start_after_res.into_string())
-            } else {
-                break;
-            }
-        }
-
-        Ok(dealers)
-    }
-
-    pub async fn get_all_nymd_epoch_dealings(
-        &self,
-        idx: usize,
-    ) -> Result<Vec<ContractDealing>, ValidatorClientError>
-    where
-        C: CosmWasmClient + Sync + Send,
-    {
-        let mut dealings = Vec::new();
-        let mut start_after = None;
-        loop {
-            let mut paged_response = self
-                .nymd
-                .get_dealings_paged(idx, start_after.take(), self.dealers_page_limit)
-                .await?;
-            dealings.append(&mut paged_response.dealings);
-
-            if let Some(start_after_res) = paged_response.start_next_after {
-                start_after = Some(start_after_res.into_string())
-            } else {
-                break;
-            }
-        }
-
-        Ok(dealings)
-    }
-
-    pub async fn get_all_nymd_verification_key_shares(
-        &self,
-    ) -> Result<Vec<ContractVKShare>, ValidatorClientError>
-    where
-        C: CosmWasmClient + Sync + Send,
-    {
-        let mut shares = Vec::new();
-        let mut start_after = None;
-        loop {
-            let mut paged_response = self
-                .nymd
-                .get_vk_shares_paged(start_after.take(), self.verification_key_page_limit)
-                .await?;
-            shares.append(&mut paged_response.shares);
-
-            if let Some(start_after_res) = paged_response.start_next_after {
-                start_after = Some(start_after_res.into_string())
-            } else {
-                break;
-            }
-        }
-
-        Ok(shares)
-    }
-
-    pub async fn get_all_nymd_proposals(
-        &self,
-    ) -> Result<Vec<ProposalResponse>, ValidatorClientError>
-    where
-        C: CosmWasmClient + Sync + Send,
-    {
-        let mut proposals = Vec::new();
-        let mut start_after = None;
-
-        loop {
-            let mut paged_response = self
-                .nymd
-                .list_proposals(start_after.take(), self.proposals_page_limit)
-                .await?;
-
-            let last_id = paged_response.proposals.last().map(|prop| prop.id);
-            proposals.append(&mut paged_response.proposals);
-
-            if let Some(start_after_res) = last_id {
-                start_after = Some(start_after_res)
-            } else {
-                break;
-            }
-        }
-
-        Ok(proposals)
-    }
 }
 
 // validator-api wrappers
@@ -727,53 +572,14 @@ impl<C> Client<C> {
     ) -> Result<BlindedSignatureResponse, ValidatorClientError> {
         Ok(self.validator_api.blind_sign(request_body).await?)
     }
-}
 
-#[derive(Clone)]
+    pub async fn get_coconut_verification_key(
-pub struct CoconutApiClient {
+        &self,
-    pub api_client: ApiClient,
+    ) -> Result<VerificationKeyResponse, ValidatorClientError> {
-    pub verification_key: VerificationKey,
+        Ok(self.validator_api.get_coconut_verification_key().await?)
-    pub node_id: NodeIndex,
-    #[cfg(feature = "nymd-client")]
-    pub cosmos_address: cosmrs::AccountId,
-}
-
-#[cfg(feature = "nymd-client")]
-impl CoconutApiClient {
-    pub async fn all_coconut_api_clients<C>(
-        nymd_client: &Client<C>,
-    ) -> Result<Vec<Self>, ValidatorClientError>
-    where
-        C: CosmWasmClient + Sync + Send,
-    {
-        Ok(nymd_client
-            .get_all_nymd_verification_key_shares()
-            .await?
-            .into_iter()
-            .filter_map(Self::try_from)
-            .collect())
-    }
-
-    fn try_from(share: ContractVKShare) -> Option<Self> {
-        if share.verified {
-            if let Ok(url_address) = Url::parse(&share.announce_address) {
-                if let Ok(verification_key) = VerificationKey::try_from_bs58(&share.share) {
-                    if let Ok(cosmos_address) = cosmrs::AccountId::from_str(share.owner.as_str()) {
-                        return Some(CoconutApiClient {
-                            api_client: ApiClient::new(url_address),
-                            verification_key,
-                            node_id: share.node_index,
-                            cosmos_address,
-                        });
-                    }
-                }
-            }
-        }
-        None
     }
 }
 
-#[derive(Clone)]
 pub struct ApiClient {
     pub validator_api: validator_api::Client,
     // TODO: perhaps if we really need it at some (currently I don't see any reasons for it)
@@ -879,6 +685,16 @@ impl ApiClient {
             .await?)
     }
 
+    pub async fn get_coconut_verification_key(
+        &self,
+    ) -> Result<VerificationKeyResponse, ValidatorClientError> {
+        Ok(self.validator_api.get_coconut_verification_key().await?)
+    }
+
+    pub async fn get_cosmos_address(&self) -> Result<CosmosAddressResponse, ValidatorClientError> {
+        Ok(self.validator_api.get_cosmos_address().await?)
+    }
+
     pub async fn verify_bandwidth_credential(
         &self,
         request_body: &VerifyCredentialBody,

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

@@ -9,8 +9,7 @@ mod error;
 pub mod nymd;
 pub mod validator_api;
 
-#[cfg(feature = "nymd-client")]
+pub use crate::client::ApiClient;
-pub use crate::client::{ApiClient, CoconutApiClient};
 pub use crate::error::ValidatorClientError;
 pub use validator_api_requests::*;
 

common/client-libs/validator-client/src/nymd/cosmwasm_client/logs.rs

@@ -6,9 +6,6 @@ use cosmrs::tendermint::abci;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 
-pub use coconut_bandwidth_contract_common::event_attributes::*;
-pub use coconut_dkg_common::event_attributes::*;
-
 // it seems that currently validators just emit stringified events (which are also returned as part of deliverTx response)
 // as theirs logs
 #[derive(Debug, Serialize, Deserialize)]

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

@@ -65,7 +65,6 @@ pub struct Config {
     pub(crate) bandwidth_claim_contract_address: Option<AccountId>,
     pub(crate) coconut_bandwidth_contract_address: Option<AccountId>,
     pub(crate) multisig_contract_address: Option<AccountId>,
-    pub(crate) coconut_dkg_contract_address: Option<AccountId>,
     // TODO: add this in later commits
     // pub(crate) gas_price: GasPrice,
 }
@@ -119,10 +118,6 @@ impl Config {
                 details.contracts.multisig_contract_address.as_ref(),
                 prefix,
             )?,
-            coconut_dkg_contract_address: Self::parse_optional_account(
-                details.contracts.coconut_dkg_contract_address.as_ref(),
-                prefix,
-            )?,
         })
     }
 }
@@ -280,14 +275,6 @@ impl<C> NymdClient<C> {
         self.config.multisig_contract_address.as_ref().unwrap()
     }
 
-    // TODO: this should get changed into Result<&AccountId, NymdError> (or Option<&AccountId> in future commits
-    // note: what unwrap is doing here is just moving a failure that would have normally
-    // occurred in `connect` when attempting to parse an empty address,
-    // so it's not introducing new source of failure (just moves it)
-    pub fn coconut_dkg_contract_address(&self) -> &AccountId {
-        self.config.coconut_dkg_contract_address.as_ref().unwrap()
-    }
-
     pub fn set_simulated_gas_multiplier(&mut self, multiplier: f32) {
         self.simulated_gas_multiplier = multiplier;
     }

common/client-libs/validator-client/src/nymd/traits/dkg_query_client.rs

@@ -1,126 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use crate::nymd::error::NymdError;
-use crate::nymd::{CosmWasmClient, NymdClient};
-use async_trait::async_trait;
-use coconut_dkg_common::dealer::{
-    DealerDetailsResponse, PagedDealerResponse, PagedDealingsResponse,
-};
-use coconut_dkg_common::msg::QueryMsg as DkgQueryMsg;
-use coconut_dkg_common::types::EpochState;
-use coconut_dkg_common::verification_key::PagedVKSharesResponse;
-use cosmrs::AccountId;
-
-#[async_trait]
-pub trait DkgQueryClient {
-    async fn get_current_epoch_state(&self) -> Result<EpochState, NymdError>;
-    async fn get_dealer_details(
-        &self,
-        address: &AccountId,
-    ) -> Result<DealerDetailsResponse, NymdError>;
-    async fn get_current_dealers_paged(
-        &self,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedDealerResponse, NymdError>;
-    async fn get_past_dealers_paged(
-        &self,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedDealerResponse, NymdError>;
-
-    async fn get_dealings_paged(
-        &self,
-        idx: usize,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedDealingsResponse, NymdError>;
-    async fn get_vk_shares_paged(
-        &self,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedVKSharesResponse, NymdError>;
-}
-
-#[async_trait]
-impl<C> DkgQueryClient for NymdClient<C>
-where
-    C: CosmWasmClient + Send + Sync,
-{
-    async fn get_current_epoch_state(&self) -> Result<EpochState, NymdError> {
-        let request = DkgQueryMsg::GetCurrentEpochState {};
-        self.client
-            .query_contract_smart(self.coconut_dkg_contract_address(), &request)
-            .await
-    }
-    async fn get_dealer_details(
-        &self,
-        address: &AccountId,
-    ) -> Result<DealerDetailsResponse, NymdError> {
-        let request = DkgQueryMsg::GetDealerDetails {
-            dealer_address: address.to_string(),
-        };
-        self.client
-            .query_contract_smart(self.coconut_dkg_contract_address(), &request)
-            .await
-    }
-
-    async fn get_current_dealers_paged(
-        &self,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedDealerResponse, NymdError> {
-        let request = DkgQueryMsg::GetCurrentDealers {
-            start_after,
-            limit: page_limit,
-        };
-        self.client
-            .query_contract_smart(self.coconut_dkg_contract_address(), &request)
-            .await
-    }
-
-    async fn get_past_dealers_paged(
-        &self,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedDealerResponse, NymdError> {
-        let request = DkgQueryMsg::GetPastDealers {
-            start_after,
-            limit: page_limit,
-        };
-        self.client
-            .query_contract_smart(self.coconut_dkg_contract_address(), &request)
-            .await
-    }
-
-    async fn get_dealings_paged(
-        &self,
-        idx: usize,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedDealingsResponse, NymdError> {
-        let request = DkgQueryMsg::GetDealing {
-            idx: idx as u64,
-            limit: page_limit,
-            start_after,
-        };
-        self.client
-            .query_contract_smart(self.coconut_dkg_contract_address(), &request)
-            .await
-    }
-
-    async fn get_vk_shares_paged(
-        &self,
-        start_after: Option<String>,
-        page_limit: Option<u32>,
-    ) -> Result<PagedVKSharesResponse, NymdError> {
-        let request = DkgQueryMsg::GetVerificationKeys {
-            limit: page_limit,
-            start_after,
-        };
-        self.client
-            .query_contract_smart(self.coconut_dkg_contract_address(), &request)
-            .await
-    }
-}

common/client-libs/validator-client/src/nymd/traits/dkg_signing_client.rs

@@ -1,100 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use crate::nymd::cosmwasm_client::types::ExecuteResult;
-use crate::nymd::error::NymdError;
-use crate::nymd::{Fee, NymdClient, SigningCosmWasmClient};
-use async_trait::async_trait;
-use coconut_dkg_common::msg::ExecuteMsg as DkgExecuteMsg;
-use coconut_dkg_common::types::EncodedBTEPublicKeyWithProof;
-use coconut_dkg_common::verification_key::VerificationKeyShare;
-use contracts_common::dealings::ContractSafeBytes;
-
-#[async_trait]
-pub trait DkgSigningClient {
-    async fn register_dealer(
-        &self,
-        bte_key: EncodedBTEPublicKeyWithProof,
-        announce_address: String,
-        fee: Option<Fee>,
-    ) -> Result<ExecuteResult, NymdError>;
-
-    async fn submit_dealing_bytes(
-        &self,
-        commitment: ContractSafeBytes,
-        fee: Option<Fee>,
-    ) -> Result<ExecuteResult, NymdError>;
-
-    async fn submit_verification_key_share(
-        &self,
-        share: VerificationKeyShare,
-        fee: Option<Fee>,
-    ) -> Result<ExecuteResult, NymdError>;
-}
-
-#[async_trait]
-impl<C> DkgSigningClient for NymdClient<C>
-where
-    C: SigningCosmWasmClient + Send + Sync,
-{
-    async fn register_dealer(
-        &self,
-        bte_key: EncodedBTEPublicKeyWithProof,
-        announce_address: String,
-        fee: Option<Fee>,
-    ) -> Result<ExecuteResult, NymdError> {
-        let req = DkgExecuteMsg::RegisterDealer {
-            bte_key_with_proof: bte_key,
-            announce_address,
-        };
-
-        self.client
-            .execute(
-                self.address(),
-                self.coconut_dkg_contract_address(),
-                &req,
-                fee.unwrap_or_default(),
-                format!("registering {} as a dealer", self.address()),
-                vec![],
-            )
-            .await
-    }
-
-    async fn submit_dealing_bytes(
-        &self,
-        dealing_bytes: ContractSafeBytes,
-        fee: Option<Fee>,
-    ) -> Result<ExecuteResult, NymdError> {
-        let req = DkgExecuteMsg::CommitDealing { dealing_bytes };
-
-        self.client
-            .execute(
-                self.address(),
-                self.coconut_dkg_contract_address(),
-                &req,
-                fee.unwrap_or_default(),
-                "dealing commitment",
-                vec![],
-            )
-            .await
-    }
-
-    async fn submit_verification_key_share(
-        &self,
-        share: VerificationKeyShare,
-        fee: Option<Fee>,
-    ) -> Result<ExecuteResult, NymdError> {
-        let req = DkgExecuteMsg::CommitVerificationKeyShare { share };
-
-        self.client
-            .execute(
-                self.address(),
-                self.coconut_dkg_contract_address(),
-                &req,
-                fee.unwrap_or_default(),
-                "verification key share commitment",
-                vec![],
-            )
-            .await
-    }
-}

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

@@ -3,8 +3,6 @@
 
 mod coconut_bandwidth_query_client;
 mod coconut_bandwidth_signing_client;
-mod dkg_query_client;
-mod dkg_signing_client;
 mod mixnet_query_client;
 mod mixnet_signing_client;
 mod multisig_query_client;
@@ -14,8 +12,6 @@ mod vesting_signing_client;
 
 pub use coconut_bandwidth_query_client::CoconutBandwidthQueryClient;
 pub use coconut_bandwidth_signing_client::CoconutBandwidthSigningClient;
-pub use dkg_query_client::DkgQueryClient;
-pub use dkg_signing_client::DkgSigningClient;
 pub use mixnet_query_client::MixnetQueryClient;
 pub use mixnet_signing_client::MixnetSigningClient;
 pub use multisig_query_client::MultisigQueryClient;

common/client-libs/validator-client/src/nymd/traits/multisig_query_client.rs

@@ -4,19 +4,13 @@
 use crate::nymd::error::NymdError;
 use crate::nymd::{CosmWasmClient, NymdClient};
 
-use cw3::{ProposalListResponse, ProposalResponse};
+use multisig_contract_common::msg::{ProposalResponse, QueryMsg};
-use multisig_contract_common::msg::QueryMsg;
 
 use async_trait::async_trait;
 
 #[async_trait]
 pub trait MultisigQueryClient {
     async fn get_proposal(&self, proposal_id: u64) -> Result<ProposalResponse, NymdError>;
-    async fn list_proposals(
-        &self,
-        start_after: Option<u64>,
-        limit: Option<u32>,
-    ) -> Result<ProposalListResponse, NymdError>;
 }
 
 #[async_trait]
@@ -27,15 +21,4 @@ impl<C: CosmWasmClient + Sync + Send> MultisigQueryClient for NymdClient<C> {
             .query_contract_smart(self.multisig_contract_address(), &request)
             .await
     }
-
-    async fn list_proposals(
-        &self,
-        start_after: Option<u64>,
-        limit: Option<u32>,
-    ) -> Result<ProposalListResponse, NymdError> {
-        let request = QueryMsg::ListProposals { start_after, limit };
-        self.client
-            .query_contract_smart(self.multisig_contract_address(), &request)
-            .await
-    }
 }

common/client-libs/validator-client/src/nymd/traits/multisig_signing_client.rs

@@ -7,11 +7,11 @@ use crate::nymd::error::NymdError;
 use crate::nymd::{Fee, NymdClient};
 
 use coconut_bandwidth_contract_common::msg::ExecuteMsg as CoconutBandwidthExecuteMsg;
-use cw3::Vote;
 use multisig_contract_common::msg::ExecuteMsg;
 
 use async_trait::async_trait;
 use cosmwasm_std::{to_binary, Coin, CosmosMsg, WasmMsg};
+use cw3::Vote;
 
 #[async_trait]
 pub trait MultisigSigningClient {

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

@@ -9,7 +9,8 @@ use reqwest::Response;
 use serde::{Deserialize, Serialize};
 use url::Url;
 use validator_api_requests::coconut::{
-    BlindSignRequestBody, BlindedSignatureResponse, VerifyCredentialBody, VerifyCredentialResponse,
+    BlindSignRequestBody, BlindedSignatureResponse, CosmosAddressResponse, VerificationKeyResponse,
+    VerifyCredentialBody, VerifyCredentialResponse,
 };
 use validator_api_requests::models::{
     GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse,
@@ -26,7 +27,6 @@ type Params<'a, K, V> = &'a [(K, V)];
 
 const NO_PARAMS: Params<'_, &'_ str, &'_ str> = &[];
 
-#[derive(Clone)]
 pub struct Client {
     url: Url,
     reqwest_client: reqwest::Client,
@@ -443,6 +443,34 @@ impl Client {
         .await
     }
 
+    pub async fn get_coconut_verification_key(
+        &self,
+    ) -> Result<VerificationKeyResponse, ValidatorAPIError> {
+        self.query_validator_api(
+            &[
+                routes::API_VERSION,
+                routes::COCONUT_ROUTES,
+                routes::BANDWIDTH,
+                routes::COCONUT_VERIFICATION_KEY,
+            ],
+            NO_PARAMS,
+        )
+        .await
+    }
+
+    pub async fn get_cosmos_address(&self) -> Result<CosmosAddressResponse, ValidatorAPIError> {
+        self.query_validator_api(
+            &[
+                routes::API_VERSION,
+                routes::COCONUT_ROUTES,
+                routes::BANDWIDTH,
+                routes::COCONUT_COSMOS_ADDRESS,
+            ],
+            NO_PARAMS,
+        )
+        .await
+    }
+
     pub async fn verify_bandwidth_credential(
         &self,
         request_body: &VerifyCredentialBody,

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

@@ -15,6 +15,8 @@ pub const BANDWIDTH: &str = "bandwidth";
 
 pub const COCONUT_BLIND_SIGN: &str = "blind-sign";
 pub const COCONUT_PARTIAL_BANDWIDTH_CREDENTIAL: &str = "partial-bandwidth-credential";
+pub const COCONUT_VERIFICATION_KEY: &str = "verification-key";
+pub const COCONUT_COSMOS_ADDRESS: &str = "cosmos-address";
 pub const COCONUT_VERIFY_BANDWIDTH_CREDENTIAL: &str = "verify-bandwidth-credential";
 
 pub const STATUS_ROUTES: &str = "status";

common/commands/Cargo.toml

@@ -22,7 +22,6 @@ thiserror = "1"
 time = { version = "0.3.6", features = ["parsing", "formatting"] }
 toml = "0.5.6"
 url = "2.2"
-tap = "1"
 
 cosmrs = { git = "https://github.com/neacsu/cosmos-rust", branch = "neacsu/feegrant_support" }
 cosmwasm-std = { version = "1.0.0" }

common/commands/src/context/errors.rs

@@ -15,10 +15,4 @@ pub enum ContextError {
     // TODO: improve this to return known errors
     #[error("failed to create client - {0}")]
     NymdError(String),
-
-    #[error("{0}")]
-    NymdErrorPassthrough(#[from] validator_client::nymd::error::NymdError),
-
-    #[error("{0}")]
-    ValidatorClientError(#[from] validator_client::ValidatorClientError),
 }

common/commands/src/context/mod.rs

@@ -6,7 +6,6 @@ use network_defaults::{
     var_names::{API_VALIDATOR, MIXNET_CONTRACT_ADDRESS, NYMD_VALIDATOR, VESTING_CONTRACT_ADDRESS},
     NymNetworkDetails,
 };
-use tap::prelude::*;
 use validator_client::nymd::{self, AccountId, NymdClient, QueryNymdClient, SigningNymdClient};
 pub use validator_client::validator_api::Client as ValidatorApiClient;
 
@@ -59,7 +58,7 @@ pub fn create_signing_client(
     network_details: &NymNetworkDetails,
 ) -> Result<SigningClient, ContextError> {
     let client_config = nymd::Config::try_from_nym_network_details(network_details)
-        .tap_err(|e| log::error!("Failed to get client config - {:?}", e))?;
+        .expect("failed to construct valid validator client config with the provided network");
 
     // get mnemonic
     let mnemonic = match std::env::var("MNEMONIC") {
@@ -88,7 +87,7 @@ pub fn create_query_client(
     network_details: &NymNetworkDetails,
 ) -> Result<QueryClient, ContextError> {
     let client_config = nymd::Config::try_from_nym_network_details(network_details)
-        .tap_err(|e| log::error!("Failed to get client config - {:?}", e))?;
+        .expect("failed to construct valid validator client config with the provided network");
 
     let nymd_url = network_details
         .endpoints
@@ -108,7 +107,7 @@ pub fn create_signing_client_with_validator_api(
     network_details: &NymNetworkDetails,
 ) -> Result<SigningClientWithValidatorAPI, ContextError> {
     let client_config = validator_client::Config::try_from_nym_network_details(network_details)
-        .tap_err(|e| log::error!("Failed to get client config - {:?}", e))?;
+        .expect("failed to construct valid validator client config with the provided network");
 
     // get mnemonic
     let mnemonic = match std::env::var("MNEMONIC") {
@@ -130,7 +129,7 @@ pub fn create_query_client_with_validator_api(
     network_details: &NymNetworkDetails,
 ) -> Result<QueryClientWithValidatorAPI, ContextError> {
     let client_config = validator_client::Config::try_from_nym_network_details(network_details)
-        .tap_err(|e| log::error!("Failed to get client config - {:?}", e))?;
+        .expect("failed to construct valid validator client config with the provided network");
 
     match validator_client::client::Client::new_query(client_config) {
         Ok(client) => Ok(client),

common/commands/src/validator/signature/sign.rs

@@ -13,7 +13,7 @@ use validator_client::nymd::wallet::DirectSecp256k1HdWallet;
 pub struct SignatureOutputJson {
     pub account_id: String,
     pub public_key: PublicKey,
-    pub signature_as_hex: String,
+    pub signature: String,
 }
 
 #[derive(Debug, Parser)]
@@ -46,7 +46,7 @@ pub fn sign(args: Args, prefix: &str, mnemonic: Option<bip39::Mnemonic>) {
                             let output = SignatureOutputJson {
                                 account_id: account.address().to_string(),
                                 public_key: account.public_key(),
-                                signature_as_hex: signature.to_string(),
+                                signature: signature.to_string(),
                             };
                             println!("{}", json!(output));
                         }

common/commands/src/validator/vesting/balance.rs

@@ -3,11 +3,11 @@
 
 use clap::Parser;
 use cosmrs::AccountId;
-use log::{error, info};
+use log::info;
 
 use validator_client::nymd::{Coin, VestingQueryClient};
 
-use crate::context::QueryClient;
+use crate::context::SigningClient;
 use crate::utils::show_error;
 use crate::utils::{pretty_coin, pretty_cosmwasm_coin};
 
@@ -18,16 +18,8 @@ pub struct Args {
     pub address: Option<AccountId>,
 }
 
-pub async fn balance(args: Args, client: QueryClient, address_from_mnemonic: Option<AccountId>) {
+pub async fn balance(args: Args, client: SigningClient) {
-    if args.address.is_none() && address_from_mnemonic.is_none() {
+    let account_id = args.address.unwrap_or_else(|| client.address().clone());
-        error!("Please specify an account address or a mnemonic to get the balance for");
-        return;
-    }
-
-    let account_id = args
-        .address
-        .unwrap_or_else(|| address_from_mnemonic.expect("please provide a mnemonic"));
-
     let vesting_address = account_id.to_string();
     let denom = client.current_chain_details().mix_denom.base.as_str();
 

common/commands/src/validator/vesting/query_vesting_schedule.rs

@@ -4,11 +4,11 @@
 use clap::Parser;
 use cosmrs::AccountId;
 use cosmwasm_std::Coin as CosmWasmCoin;
-use log::{error, info};
+use log::info;
 
 use validator_client::nymd::{Coin, VestingQueryClient};
 
-use crate::context::QueryClient;
+use crate::context::SigningClient;
 use crate::utils::show_error;
 use crate::utils::{pretty_coin, pretty_cosmwasm_coin};
 
@@ -19,18 +19,8 @@ pub struct Args {
     pub address: Option<AccountId>,
 }
 
-pub async fn query(args: Args, client: QueryClient, address_from_mnemonic: Option<AccountId>) {
+pub async fn query(args: Args, client: SigningClient) {
-    if args.address.is_none() && address_from_mnemonic.is_none() {
+    let account_id = args.address.unwrap_or_else(|| client.address().clone());
-        error!("Please specify an account address or a mnemonic to get the balance for");
-        return;
-    }
-
-    let account_id = args
-        .address
-        .unwrap_or_else(|| address_from_mnemonic.expect("please provide a mnemonic"));
-
-    info!("Checking account {} for a vesting schedule...", account_id);
-
     let vesting_address = account_id.to_string();
     let denom = client.current_chain_details().mix_denom.base.as_str();
 

common/cosmwasm-smart-contracts/coconut-bandwidth-contract/src/event_attributes.rs

@@ -1,4 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-pub const BANDWIDTH_PROPOSAL_ID: &str = "proposal_id";

common/cosmwasm-smart-contracts/coconut-bandwidth-contract/src/lib.rs

@@ -1,5 +1,4 @@
 pub mod deposit;
-pub mod event_attributes;
 pub mod events;
 pub mod msg;
 pub mod spend_credential;

common/cosmwasm-smart-contracts/coconut-dkg/Cargo.toml

@@ -1,15 +0,0 @@
-[package]
-name = "coconut-dkg-common"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-cosmwasm-std = "1.0.0"
-cw-utils = "0.13.4"
-schemars = "0.8"
-serde = { version = "1.0.103", default-features = false, features = ["derive"] }
-
-contracts-common = { path = "../contracts-common" }
-multisig-contract-common = { path = "../multisig-contract" }

common/cosmwasm-smart-contracts/coconut-dkg/src/dealer.rs

@@ -1,102 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use crate::types::{ContractSafeBytes, EncodedBTEPublicKeyWithProof, NodeIndex};
-use cosmwasm_std::Addr;
-use serde::{Deserialize, Serialize};
-
-#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq)]
-#[serde(rename_all = "snake_case")]
-pub struct DealerDetails {
-    pub address: Addr,
-    pub bte_public_key_with_proof: EncodedBTEPublicKeyWithProof,
-    pub announce_address: String,
-    pub assigned_index: NodeIndex,
-}
-
-#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub enum DealerType {
-    Current,
-    Past,
-    Unknown,
-}
-
-impl DealerType {
-    pub fn is_current(&self) -> bool {
-        matches!(&self, DealerType::Current)
-    }
-}
-
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub struct DealerDetailsResponse {
-    pub details: Option<DealerDetails>,
-    pub dealer_type: DealerType,
-}
-
-impl DealerDetailsResponse {
-    pub fn new(details: Option<DealerDetails>, dealer_type: DealerType) -> Self {
-        DealerDetailsResponse {
-            details,
-            dealer_type,
-        }
-    }
-}
-
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub struct PagedDealerResponse {
-    pub dealers: Vec<DealerDetails>,
-    pub per_page: usize,
-    pub start_next_after: Option<Addr>,
-}
-
-impl PagedDealerResponse {
-    pub fn new(
-        dealers: Vec<DealerDetails>,
-        per_page: usize,
-        start_next_after: Option<Addr>,
-    ) -> Self {
-        PagedDealerResponse {
-            dealers,
-            per_page,
-            start_next_after,
-        }
-    }
-}
-
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub struct ContractDealing {
-    pub dealing: ContractSafeBytes,
-    pub dealer: Addr,
-}
-
-impl ContractDealing {
-    pub fn new(dealing: ContractSafeBytes, dealer: Addr) -> Self {
-        ContractDealing { dealing, dealer }
-    }
-}
-
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub struct PagedDealingsResponse {
-    pub dealings: Vec<ContractDealing>,
-    pub per_page: usize,
-    pub start_next_after: Option<Addr>,
-}
-
-impl PagedDealingsResponse {
-    pub fn new(
-        dealings: Vec<ContractDealing>,
-        per_page: usize,
-        start_next_after: Option<Addr>,
-    ) -> Self {
-        PagedDealingsResponse {
-            dealings,
-            per_page,
-            start_next_after,
-        }
-    }
-}

common/cosmwasm-smart-contracts/coconut-dkg/src/event_attributes.rs

@@ -1,5 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-pub const NODE_INDEX: &str = "node_index";
-pub const DKG_PROPOSAL_ID: &str = "proposal_id";

common/cosmwasm-smart-contracts/coconut-dkg/src/lib.rs

@@ -1,5 +0,0 @@
-pub mod dealer;
-pub mod event_attributes;
-pub mod msg;
-pub mod types;
-pub mod verification_key;

common/cosmwasm-smart-contracts/coconut-dkg/src/msg.rs

@@ -1,69 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use crate::types::{ContractSafeBytes, EncodedBTEPublicKeyWithProof};
-use crate::verification_key::VerificationKeyShare;
-use cosmwasm_std::Addr;
-use schemars::JsonSchema;
-use serde::{Deserialize, Serialize};
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
-pub struct InstantiateMsg {
-    pub group_addr: String,
-    pub multisig_addr: String,
-    pub admin: String,
-    pub mix_denom: String,
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, JsonSchema)]
-#[serde(rename_all = "snake_case")]
-pub enum ExecuteMsg {
-    RegisterDealer {
-        bte_key_with_proof: EncodedBTEPublicKeyWithProof,
-        announce_address: String,
-    },
-
-    CommitDealing {
-        dealing_bytes: ContractSafeBytes,
-    },
-
-    CommitVerificationKeyShare {
-        share: VerificationKeyShare,
-    },
-
-    VerifyVerificationKeyShare {
-        owner: Addr,
-    },
-
-    AdvanceEpochState {},
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
-#[serde(rename_all = "snake_case")]
-pub enum QueryMsg {
-    GetCurrentEpochState {},
-    GetDealerDetails {
-        dealer_address: String,
-    },
-    GetCurrentDealers {
-        limit: Option<u32>,
-        start_after: Option<String>,
-    },
-    GetPastDealers {
-        limit: Option<u32>,
-        start_after: Option<String>,
-    },
-    GetDealing {
-        idx: u64,
-        limit: Option<u32>,
-        start_after: Option<String>,
-    },
-    GetVerificationKeys {
-        limit: Option<u32>,
-        start_after: Option<String>,
-    },
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
-#[serde(rename_all = "snake_case")]
-pub struct MigrateMsg {}

common/cosmwasm-smart-contracts/coconut-dkg/src/types.rs

@@ -1,81 +0,0 @@
-// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use serde::{Deserialize, Serialize};
-use std::fmt::{Display, Formatter};
-
-pub use crate::dealer::{DealerDetails, PagedDealerResponse};
-pub use contracts_common::dealings::ContractSafeBytes;
-pub use cosmwasm_std::{Addr, Coin};
-
-pub type EncodedBTEPublicKeyWithProof = String;
-pub type EncodedBTEPublicKeyWithProofRef<'a> = &'a str;
-pub type NodeIndex = u64;
-
-// 2 public attributes, 2 private attributes, 1 fixed for coconut credential
-pub const TOTAL_DEALINGS: usize = 2 + 2 + 1;
-
-// currently (it is still extremely likely to change, we might be able to get rid of verification key-related complaints),
-// the epoch can be in the following states (in order):
-// 1. PublicKeySubmission -> potential dealers are submitting their BTE and ed25519 public keys to participate in dealing exchange
-// 2. DealingExchange -> the actual (off-chain) dealing exchange is happening
-// 3. ComplaintSubmission -> receivers submitting evidence of other dealers sending malformed data
-// 4. ComplaintVoting -> (if any complaints were submitted) receivers voting on the validity of the evidence provided
-// 5. VerificationKeySubmission -> receivers submitting their partial (and master) verification keys
-// 6. VerificationKeyMismatchSubmission -> receivers / watchers raising issue that the submitted VK are mismatched with their local derivations
-// 7. VerificationKeyMismatchVoting -> (if any complaints were submitted) receivers voting on received mismatches
-// 8. InProgress -> all receivers have all their secrets derived and all is good
-//
-// Note: It's important that the variant ordering is not changed otherwise it would mess up the derived `PartialOrd`
-#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Ord, PartialOrd)]
-#[serde(rename_all = "snake_case")]
-pub enum EpochState {
-    PublicKeySubmission,
-    DealingExchange,
-    VerificationKeySubmission,
-    VerificationKeyValidation,
-    VerificationKeyFinalization,
-    InProgress,
-}
-
-impl Default for EpochState {
-    fn default() -> Self {
-        Self::PublicKeySubmission
-    }
-}
-
-impl Display for EpochState {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        match self {
-            EpochState::PublicKeySubmission => write!(f, "PublicKeySubmission"),
-            EpochState::DealingExchange => write!(f, "DealingExchange"),
-            EpochState::VerificationKeySubmission => write!(f, "VerificationKeySubmission"),
-            EpochState::VerificationKeyValidation => write!(f, "VerificationKeyValidation"),
-            EpochState::VerificationKeyFinalization => write!(f, "VerificationKeyFinalization"),
-            EpochState::InProgress => write!(f, "InProgress"),
-        }
-    }
-}
-
-impl EpochState {
-    pub fn next(self) -> Option<Self> {
-        match self {
-            EpochState::PublicKeySubmission => Some(EpochState::DealingExchange),
-            EpochState::DealingExchange => Some(EpochState::VerificationKeySubmission),
-            EpochState::VerificationKeySubmission => Some(EpochState::VerificationKeyValidation),
-            EpochState::VerificationKeyValidation => Some(EpochState::VerificationKeyFinalization),
-            EpochState::VerificationKeyFinalization => Some(EpochState::InProgress),
-            EpochState::InProgress => None,
-        }
-    }
-
-    pub fn all_until(&self, end: Self) -> Vec<Self> {
-        let mut states = vec![*self];
-        while states.last().unwrap() != &end {
-            let next_state = states.last().unwrap().next().expect("somehow reached the end of state diff -> this should be impossible under any circumstances!");
-            states.push(next_state);
-        }
-
-        states
-    }
-}