Merge branch 'release/v1.1.2'

* Expose name of standard directories

* Use one command instead of two

.github/workflows/audit.yml

@@ -1,36 +1,52 @@
 name: Daily security audit
 
-on: workflow_dispatch
+on:
+  schedule:
+    - cron: '5 9 * * *'
 jobs:
-  security_audit:
-    runs-on: ubuntu-latest
+  cargo-deny:
+    runs-on: ubuntu-20.04
     steps:
-      - uses: actions/checkout@v1
-      - uses: actions-rs/audit-check@v1
+      - name: Checkout repository code
+        uses: actions/checkout@v2
+      - name: Install rust toolchain
+        uses: actions-rs/toolchain@v1
         with:
-          token: ${{ secrets.GITHUB_TOKEN }}
-
+          toolchain: stable
+      - name: Install cargo deny
+        run: cargo install --locked cargo-deny
+      - name: Run cargo deny
+        run: |
+          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
+          path: .github/workflows/support-files/notifications/deny.message
   notification:
-    if: ${{ failure() }}
-    needs: security_audit
-    runs-on: ubuntu-latest
+    needs: cargo-deny
+    runs-on: ubuntu-20.04
     steps:
       - name: Check out repository code
         uses: actions/checkout@v2
+      - name: Download report from previous job
+        uses: actions/download-artifact@v3
+        with:
+          name: report
+          path: .github/workflows/support-files/notifications
       - name: Keybase - Node Install
         run: npm install
         working-directory: .github/workflows/support-files
       - name: Keybase - Send Notification
         env:
-          NYM_NOTIFICATION_KIND: nightly
-          NYM_PROJECT_NAME: "Nym daily audit"
+          NYM_NOTIFICATION_KIND: security
+          NYM_PROJECT_NAME: "Daily security report"
           GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
-          GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
-          GIT_BRANCH: "${GITHUB_REF##*/}"
           KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
           KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
-          KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBTECH_TEAM }}"
-          KEYBASE_NYM_CHANNEL: "test"
+          KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
+          KEYBASE_NYM_CHANNEL: "security"
         uses: docker://keybaseio/client:stable-node
         with:
           args: .github/workflows/support-files/notifications/entry_point.sh

.github/workflows/build.yml

@@ -29,6 +29,12 @@ 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:
@@ -48,12 +54,6 @@ 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,6 +66,8 @@ jobs:
           command: clippy
           args: --workspace -- -D warnings
 
+      # COCONUT stuff
+
       - name: Build all binaries with coconut enabled
         uses: actions-rs/cargo@v1
         with:
@@ -82,4 +84,4 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: clippy
-          args: --features=coconut -- -D warnings
+          args: --all-targets --features=coconut -- -D warnings

.github/workflows/build_matrix_includes.json

@@ -1,6 +1,6 @@
 [
   {
-    "os":"ubuntu-latest",
+    "os":"ubuntu-20.04",
     "rust":"stable",
     "runOnEvent":"always"
   },

.github/workflows/build_on_dispatch.yml

@@ -1,72 +0,0 @@
-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,16 +1,21 @@
 name: Build release of Nym smart contracts
 on:
   workflow_dispatch:
-
-defaults:
-  run:
-    working-directory: contracts
+  release:
+    types: [created]
 
 jobs:
   build:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     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
@@ -21,7 +26,7 @@ jobs:
           components: rustfmt, clippy
 
       - name: Build release contracts
-        run: RUSTFLAGS='-C link-arg=-s' cargo build --release --target wasm32-unknown-unknown
+        run: make wasm
 
       - name: Upload Mixnet Contract Artifact
         uses: actions/upload-artifact@v3
@@ -36,3 +41,11 @@ 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/contracts.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   matrix_prep:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     outputs:
       matrix: ${{ steps.set-matrix.outputs.matrix }}
     steps:
@@ -24,7 +24,7 @@ jobs:
   contracts:
     # since it's going to be compiled into wasm, there's absolutely
     # no point in running CI on different OS-es
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     continue-on-error: ${{ matrix.rust == 'nightly' }}
     needs: matrix_prep
     strategy:

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

@@ -0,0 +1,56 @@
+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-20.04]
+
+    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.yml

@@ -5,7 +5,7 @@ on:
     - cron: '14 1 * * *'
 jobs:
   matrix_prep:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     outputs:
       matrix: ${{ steps.set-matrix.outputs.matrix }}
     steps:
@@ -25,7 +25,7 @@ jobs:
     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'
+        if: matrix.os == 'ubuntu-20.04'
 
       - name: Check out repository code
         uses: actions/checkout@v2
@@ -96,7 +96,7 @@ jobs:
 
       - name: Reclaim some disk space
         uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' || matrix.os == 'ubuntu-latest' }}
+        if: ${{ matrix.os == 'windows-latest' || matrix.os == 'ubuntu-20.04' }}
         with:
           command: clean
 
@@ -160,7 +160,7 @@ jobs:
 
   notification:
     needs: build
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     steps:
       - name: Collect jobs status
         uses: technote-space/workflow-conclusion-action@v2

.github/workflows/nightly_build_matrix_includes.json

@@ -1,6 +1,6 @@
 [
   {
-    "os":"ubuntu-latest",
+    "os":"ubuntu-20.04",
     "rust":"stable",
     "runOnEvent":"schedule"
   },
@@ -17,7 +17,7 @@
   },
 
   {
-    "os":"ubuntu-latest",
+    "os":"ubuntu-20.04",
     "rust":"beta",
     "runOnEvent":"schedule"
   },
@@ -33,7 +33,7 @@
   },
 
   {
-    "os":"ubuntu-latest",
+    "os":"ubuntu-20.04",
     "rust":"nightly",
     "runOnEvent":"schedule"
   },

.github/workflows/nightly_build_matrix_on_dispatch.json

@@ -1,50 +0,0 @@
-[
-  {
-    "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_release.yml

@@ -1,32 +1,49 @@
-name: Nightly builds on dispatch
+name: Nightly builds on latest release
 
-on: workflow_dispatch
+on:
+  schedule:
+    - cron: '14 2 * * *'
 jobs:
   matrix_prep:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     outputs:
       matrix: ${{ steps.set-matrix.outputs.matrix }}
     steps:
       # creates the matrix strategy from nightly_build_matrix_includes.json
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - id: set-matrix
         uses: JoshuaTheMiller/conditional-build-matrix@main
         with:
-          inputFile: '.github/workflows/nightly_build_matrix_on_dispatch.json'
+          inputFile: '.github/workflows/nightly_build_matrix_includes.json'
           filter: '[?runOnEvent==`${{ github.event_name }}` || runOnEvent==`always`]'
-  build:
+  get_release:
+    runs-on: ubuntu-20.04
     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]
     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 squashfs-tools
-        if: matrix.os == 'ubuntu-latest'
+        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-20.04'
 
-      - name: Check out repository code
-        uses: actions/checkout@v2
+      - 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
@@ -42,6 +59,12 @@ jobs:
           command: build
           args: --workspace
 
+      - name: Reclaim some disk space (because Windows is being annoying)
+        uses: actions-rs/cargo@v1
+        if: ${{ matrix.os == 'windows-latest' }}
+        with:
+          command: clean
+
       - name: Run all tests
         uses: actions-rs/cargo@v1
         with:
@@ -88,7 +111,7 @@ jobs:
 
       - name: Reclaim some disk space
         uses: actions-rs/cargo@v1
-        if: ${{ matrix.os == 'windows-latest' || matrix.os == 'ubuntu-latest' }}
+        if: ${{ matrix.os == 'windows-latest' || matrix.os == 'ubuntu-20.04' }}
         with:
           command: clean
 
@@ -99,6 +122,12 @@ 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:
@@ -145,13 +174,13 @@ jobs:
           args: --manifest-path nym-wallet/Cargo.toml --workspace --all-targets -- -D warnings
 
   notification:
-    needs: build
-    runs-on: ubuntu-latest
+    needs: [build,get_release]
+    runs-on: ubuntu-20.04
     steps:
       - name: Collect jobs status
         uses: technote-space/workflow-conclusion-action@v2
       - name: Check out repository code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: Keybase - Node Install
         if: env.WORKFLOW_CONCLUSION == 'failure'
         run: npm install
@@ -160,14 +189,14 @@ jobs:
         if: env.WORKFLOW_CONCLUSION == 'failure'
         env:
           NYM_NOTIFICATION_KIND: nightly
-          NYM_PROJECT_NAME: "Nym nightly build"
+          NYM_PROJECT_NAME: "Nym nightly build on latest release"
           GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
           GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
-          GIT_BRANCH: "${GITHUB_REF##*/}"
+          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_NYMTECH_TEAM }}"
-          KEYBASE_NYM_CHANNEL: "${{ secrets.KEYBASE_CHANNEL_DEV_CORE_ID }}"
+          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:

.github/workflows/nightly_build_release2.yml

@@ -0,0 +1,203 @@
+name: Nightly builds on second latest release
+
+on:
+  schedule:
+    - cron: '24 2 * * *'
+jobs:
+  matrix_prep:
+    runs-on: ubuntu-20.04
+    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-20.04
+    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-20.04'
+
+      - 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-20.04' }}
+        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-20.04
+    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-cli-publish.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ubuntu-latest, windows-latest, macos-latest]
+        platform: [ubuntu-20.04, windows-latest, macos-latest]
 
     runs-on: ${{ matrix.platform }}
     steps:

.github/workflows/nym-connect-publish-ubuntu.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ubuntu-latest]
+        platform: [ubuntu-20.04]
 
     runs-on: ${{ matrix.platform }}
     steps:

.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
-#      env:
-#        NYM_NOTIFICATION_KIND: nym-connect
-#        NYM_PROJECT_NAME: "nym-connect"
-#        NYM_CI_WWW_BASE: "${{ secrets.NYM_CI_WWW_BASE }}"
-#        NYM_CI_WWW_LOCATION: "nym-connect-${{ env.GITHUB_REF_SLUG }}"
-#        GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
-#        GIT_BRANCH: "${GITHUB_REF##*/}"
-#        KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
-#        KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
-#        KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
-#        KEYBASE_NYM_CHANNEL: "ci-nym-connect"
-#        IS_SUCCESS: "${{ job.status == 'success' }}"
-#      uses: docker://keybaseio/client:stable-node
-#      with:
-#        args: .github/workflows/support-files/notifications/entry_point.sh
+    - name: Keybase - Send Notification
+      env:
+        NYM_NOTIFICATION_KIND: nym-connect
+        NYM_PROJECT_NAME: "nym-connect"
+        NYM_CI_WWW_BASE: "${{ secrets.NYM_CI_WWW_BASE }}"
+        NYM_CI_WWW_LOCATION: "nym-connect-${{ env.GITHUB_REF_SLUG }}"
+        GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
+        GIT_BRANCH: "${GITHUB_REF##*/}"
+        KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
+        KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
+        KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
+        KEYBASE_NYM_CHANNEL: "ci-nym-connect"
+        IS_SUCCESS: "${{ job.status == 'success' }}"
+      uses: docker://keybaseio/client:stable-node
+      with:
+        args: .github/workflows/support-files/notifications/entry_point.sh

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

@@ -19,12 +19,15 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ubuntu-latest]
+        platform: [ubuntu-20.04]
 
     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-binaries-`
         if: startsWith(github.ref, 'refs/tags/nym-binaries-') == false && github.event_name != 'workflow_dispatch'
         uses: actions/github-script@v3

.github/workflows/nym-wallet-publish-ubuntu.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ubuntu-latest]
+        platform: [ubuntu-20.04]
 
     runs-on: ${{ matrix.platform }}
     steps:

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

@@ -12,7 +12,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ubuntu-latest]
+        platform: [ubuntu-20.04]
     runs-on: ${{ matrix.platform }}
     steps:
       - uses: actions/checkout@v2

.github/workflows/nym_wallet.yml

@@ -12,7 +12,7 @@ defaults:
 jobs:
   test:
     name: wallet tests
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v2
 

.github/workflows/support-files/nightly/index.js

@@ -85,7 +85,7 @@ async function getMessageBody(context) {
             ...
           ],
           check_run_url: 'https://api.github.com/repos/nymtech/nym/check-runs/5182940024',
-          labels: [ 'ubuntu-latest' ],
+          labels: [ 'ubuntu-20.04' ],
           runner_id: 1,
           runner_name: 'Hosted Agent',
           runner_group_id: 2,

.github/workflows/support-files/notifications/send_message.js

@@ -3,7 +3,7 @@ require('dotenv').config();
 const Bot = require('keybase-bot');
 
 let context = {
-  kinds: ['nym-wallet', 'ts-packages', 'network-explorer', 'nightly', 'nym-connect'],
+  kinds: ['nym-wallet', 'ts-packages', 'network-explorer', 'nightly', 'nym-connect','security'],
 };
 
 /**

.github/workflows/support-files/security/index.js

@@ -0,0 +1,24 @@
+const Handlebars = require('handlebars');
+const fs = require('fs');
+const path = require('path');
+const { Octokit, App } = require('octokit');
+
+async function addToContextAndValidate(context) {
+  return
+}
+
+async function getMessageBody(context) {
+  try {
+    const source = fs
+    .readFileSync("deny.message").toString();
+    return source;
+} catch (error) {
+    console.error(error);
+}
+
+}
+
+module.exports = {
+  addToContextAndValidate,
+  getMessageBody,
+};

.github/workflows/wasm_client_build.yml

@@ -7,7 +7,7 @@ on:
 
 jobs:
   wasm:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v2
 

CHANGELOG.md

@@ -2,38 +2,83 @@
 
 Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unreleased
-
-### Added
-
-- nym-cli: added CLI tool for interacting with the Nyx blockchain and Nym mixnet smart contracts ([#1577])
-- 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
-- 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])
+## [v1.1.2]
 
 ### Changed
 
-- validator-client: made `fee` argument optional for `execute` and `execute_multiple` ([#1541])
-- socks5 client: graceful shutdown should fix error on disconnect in nym-connect ([#1591])
-- wasm-client: fixed build errors on MacOS and changed example JS code to use mainnet ([#1585])
-- 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: changed error serialization on `inclusion_probability`, `stake-saturation` and `reward-estimation` endpoints to provide more accurate information ([#1681]) 
-- 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])
+- gateway: Renamed flag from `enabled/disabled_credentials_mode` to `only-coconut-credentials`
+- "Family" feature for node families + layers
+- Initial coconut functionality including credentials and distributed key generation 
 
+## [v1.1.1](https://github.com/nymtech/nym/tree/v1.1.1) (2022-11-29)
+
+### Added
+
+- binaries: add `-c` shortform for `--config-env-file`
+- 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])
+- 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])
+
+### Fixed
+
+- 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])
+- 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])
+- validator-api: changes to internal data structures due to the mixnet contract revamp ([#1472])
+- validator-api: split epoch-operations into multiple separate transactions ([#1472])
+
+[#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
@@ -41,13 +86,13 @@ 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
 [#1721]: https://github.com/nymtech/nym/pull/1721

Cargo.lock

@@ -577,9 +577,18 @@ dependencies = [
 ]
 
 [[package]]
-name = "client-core"
-version = "1.0.1"
+name = "client-connections"
+version = "0.1.0"
 dependencies = [
+ "futures",
+ "log",
+]
+
+[[package]]
+name = "client-core"
+version = "1.1.1"
+dependencies = [
+ "client-connections",
  "config",
  "crypto",
  "dirs",
@@ -627,6 +636,18 @@ 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"
@@ -736,7 +757,9 @@ checksum = "245097e9a4535ee1e3e3931fcfcd55a796a44c643e8596ff6566d68f09b87bbc"
 name = "contracts-common"
 version = "0.1.0"
 dependencies = [
+ "bs58",
  "cosmwasm-std",
+ "dkg",
  "schemars",
  "serde",
  "serde_json",
@@ -850,9 +873,9 @@ dependencies = [
 
 [[package]]
 name = "cpufeatures"
-version = "0.2.2"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "59a6001667ab124aebae2a495118e11d30984c3a653e99d86d58971708cf5e4b"
+checksum = "28d997bd5e24a5928dd43e46dc529867e207907fe0b239c3477d924f7f2ca320"
 dependencies = [
  "libc",
 ]
@@ -900,7 +923,6 @@ dependencies = [
 name = "credential"
 version = "0.1.0"
 dependencies = [
- "async-trait",
  "bip39",
  "cfg-if 0.1.10",
  "clap 3.2.8",
@@ -912,7 +934,6 @@ dependencies = [
  "crypto",
  "network-defaults",
  "pemstore",
- "pickledb",
  "rand 0.7.3",
  "serde",
  "thiserror",
@@ -942,7 +963,6 @@ dependencies = [
  "crypto",
  "rand 0.7.3",
  "thiserror",
- "url",
  "validator-api-requests",
  "validator-client",
 ]
@@ -1221,9 +1241,9 @@ dependencies = [
 
 [[package]]
 name = "cw-utils"
-version = "0.13.2"
+version = "0.13.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "babd2c090f39d07ce5bf2556962305e795daa048ce20a93709eb591476e4a29e"
+checksum = "9dbaecb78c8e8abfd6b4258c7f4fbeb5c49a5e45ee4d910d3240ee8e1d714e1b"
 dependencies = [
  "cosmwasm-std",
  "schemars",
@@ -1232,10 +1252,22 @@ dependencies = [
 ]
 
 [[package]]
-name = "cw3"
-version = "0.13.2"
+name = "cw2"
+version = "0.13.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f871854338a54c7bb094d16ffe17212b93b146d9659dbce4c9402a9b77e240ef"
+checksum = "04cf4639517490dd36b333bbd6c4fbd92e325fd0acf4683b41753bc5eb63bfc1"
+dependencies = [
+ "cosmwasm-std",
+ "cw-storage-plus",
+ "schemars",
+ "serde",
+]
+
+[[package]]
+name = "cw3"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fe19462a7f644ba60c19d3443cb90d00c50d9b6b3b0a3a7fca93df8261af979b"
 dependencies = [
  "cosmwasm-std",
  "cw-utils",
@@ -1244,10 +1276,26 @@ dependencies = [
 ]
 
 [[package]]
-name = "cw4"
-version = "0.13.2"
+name = "cw3-fixed-multisig"
+version = "0.13.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c4476d6a7c13c46ed9ff260bd0e1cf648dc37b13f483822e1ff2a431f0f6ee52"
+checksum = "df54aa54c13f405ec4ab36b6217538bc957d439eee58f89312db05a79caf6706"
+dependencies = [
+ "cosmwasm-std",
+ "cw-storage-plus",
+ "cw-utils",
+ "cw2",
+ "cw3",
+ "schemars",
+ "serde",
+ "thiserror",
+]
+
+[[package]]
+name = "cw4"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0acc3549d5ce11c6901b3a676f2e2628684722197054d97cd0101ea174ed5cbd"
 dependencies = [
  "cosmwasm-std",
  "cw-storage-plus",
@@ -1402,6 +1450,7 @@ 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",
@@ -1453,9 +1502,9 @@ dependencies = [
 
 [[package]]
 name = "ed25519"
-version = "1.4.1"
+version = "1.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3d5c4b5e5959dc2c2b89918d8e2cc40fcdd623cef026ed09d2f0ee05199dc8e4"
+checksum = "1e9c280362032ea4203659fc489832d0204ef09f247a0506f170dafcac08c369"
 dependencies = [
  "serde",
  "signature",
@@ -1582,7 +1631,7 @@ dependencies = [
 
 [[package]]
 name = "explorer-api"
-version = "1.0.1"
+version = "1.1.1"
 dependencies = [
  "chrono",
  "clap 3.2.8",
@@ -2694,12 +2743,6 @@ dependencies = [
  "vcpkg",
 ]
 
-[[package]]
-name = "linked-hash-map"
-version = "0.5.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7fb9b38af92608140b86b693604b9ffcc5824240a484d1ecd4795bacb2fe88f3"
-
 [[package]]
 name = "lioness"
 version = "0.1.2"
@@ -2930,6 +2973,7 @@ dependencies = [
  "cosmwasm-std",
  "cw-utils",
  "cw3",
+ "cw3-fixed-multisig",
  "cw4",
  "schemars",
  "serde",
@@ -3070,7 +3114,7 @@ dependencies = [
 
 [[package]]
 name = "nym-cli"
-version = "1.0.0"
+version = "1.1.0"
 dependencies = [
  "anyhow",
  "base64",
@@ -3087,6 +3131,7 @@ dependencies = [
  "pretty_env_logger",
  "serde",
  "serde_json",
+ "tap",
  "tokio",
  "validator-client",
 ]
@@ -3112,6 +3157,7 @@ dependencies = [
  "rand 0.6.5",
  "serde",
  "serde_json",
+ "tap",
  "thiserror",
  "time 0.3.14",
  "toml",
@@ -3122,9 +3168,10 @@ dependencies = [
 
 [[package]]
 name = "nym-client"
-version = "1.0.2"
+version = "1.1.1"
 dependencies = [
  "clap 3.2.8",
+ "client-connections",
  "client-core",
  "coconut-interface",
  "completions",
@@ -3146,7 +3193,9 @@ dependencies = [
  "serde",
  "serde_json",
  "sled",
+ "tap",
  "task",
+ "thiserror",
  "tokio",
  "tokio-tungstenite 0.14.0",
  "topology",
@@ -3159,7 +3208,7 @@ dependencies = [
 
 [[package]]
 name = "nym-gateway"
-version = "1.0.2"
+version = "1.1.1"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3206,7 +3255,7 @@ dependencies = [
 
 [[package]]
 name = "nym-mixnode"
-version = "1.0.2"
+version = "1.1.1"
 dependencies = [
  "anyhow",
  "bs58",
@@ -3248,10 +3297,11 @@ dependencies = [
 
 [[package]]
 name = "nym-network-requester"
-version = "1.0.2"
+version = "1.1.1"
 dependencies = [
  "async-trait",
  "clap 3.2.8",
+ "client-connections",
  "completions",
  "dirs",
  "futures",
@@ -3295,9 +3345,10 @@ dependencies = [
 
 [[package]]
 name = "nym-socks5-client"
-version = "1.0.2"
+version = "1.1.1"
 dependencies = [
  "clap 3.2.8",
+ "client-connections",
  "client-core",
  "coconut-interface",
  "completions",
@@ -3322,7 +3373,9 @@ dependencies = [
  "serde",
  "snafu 0.6.10",
  "socks5-requests",
+ "tap",
  "task",
+ "thiserror",
  "tokio",
  "topology",
  "url",
@@ -3359,13 +3412,15 @@ dependencies = [
 
 [[package]]
 name = "nym-validator-api"
-version = "1.0.2"
+version = "1.1.1"
 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",
@@ -3377,6 +3432,7 @@ dependencies = [
  "cw-utils",
  "cw3",
  "dirs",
+ "dkg",
  "dotenv",
  "futures",
  "gateway-client",
@@ -3390,6 +3446,7 @@ dependencies = [
  "nymcoconut",
  "nymsphinx",
  "okapi",
+ "pemstore",
  "pin-project",
  "pretty_env_logger",
  "rand 0.7.3",
@@ -3442,16 +3499,19 @@ name = "nymcoconut"
 version = "0.5.0"
 dependencies = [
  "bincode",
- "bls12_381 0.5.0",
+ "bls12_381 0.6.0",
  "bs58",
  "criterion",
  "digest 0.9.0",
+ "dkg",
  "doc-comment",
- "ff 0.10.1",
+ "ff 0.11.0",
  "getrandom 0.2.6",
- "group 0.10.0",
+ "group 0.11.0",
  "itertools",
+ "pemstore",
  "rand 0.8.5",
+ "rand_chacha 0.3.1",
  "serde",
  "serde_derive",
  "sha2 0.9.9",
@@ -3866,19 +3926,6 @@ dependencies = [
  "indexmap",
 ]
 
-[[package]]
-name = "pickledb"
-version = "0.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9161694d67f6c5163519d42be942ae36bbdb55f439460144f105bc4f9f7d1d61"
-dependencies = [
- "bincode",
- "serde",
- "serde_cbor",
- "serde_json",
- "serde_yaml",
-]
-
 [[package]]
 name = "pin-project"
 version = "1.0.10"
@@ -4121,6 +4168,7 @@ name = "proxy-helpers"
 version = "0.1.0"
 dependencies = [
  "bytes",
+ "client-connections",
  "futures",
  "log",
  "ordered-buffer",
@@ -4988,18 +5036,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "serde_yaml"
-version = "0.8.23"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4a521f2940385c165a24ee286aa8599633d162077a54bdcae2a6fd5a7bfa7a0"
-dependencies = [
- "indexmap",
- "ryu",
- "serde",
- "yaml-rust",
-]
-
 [[package]]
 name = "sha-1"
 version = "0.8.2"
@@ -5626,7 +5662,9 @@ checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369"
 name = "task"
 version = "0.1.0"
 dependencies = [
+ "futures",
  "log",
+ "thiserror",
  "tokio",
 ]
 
@@ -5766,18 +5804,18 @@ checksum = "b1141d4d61095b28419e22cb0bbf02755f5e54e0526f97f1e3d1d160e60885fb"
 
 [[package]]
 name = "thiserror"
-version = "1.0.35"
+version = "1.0.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c53f98874615aea268107765aa1ed8f6116782501d18e53d08b471733bea6c85"
+checksum = "10deb33631e3c9018b9baf9dcbbc4f737320d2b576bac10f6aefa048fa407e3e"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "1.0.35"
+version = "1.0.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8b463991b4eab2d801e724172285ec4195c650e8ec79b149e6c2a8e6dd3f783"
+checksum = "982d17546b47146b28f7c22e3d08465f6b8903d0ea13c1660d9d84a6e7adcdbb"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -6405,6 +6443,7 @@ dependencies = [
  "base64",
  "bip39",
  "coconut-bandwidth-contract-common",
+ "coconut-dkg-common",
  "coconut-interface",
  "colored",
  "config",
@@ -6514,7 +6553,6 @@ version = "0.1.0"
 dependencies = [
  "contracts-common",
  "cosmwasm-std",
- "log",
  "mixnet-contract-common",
  "schemars",
  "serde",
@@ -6867,15 +6905,6 @@ dependencies = [
  "zeroize",
 ]
 
-[[package]]
-name = "yaml-rust"
-version = "0.4.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56c1936c4cc7a1c9ab21a1ebb602eb942ba868cbd44a99cb7cdc5892335e1c85"
-dependencies = [
- "linked-hash-map",
-]
-
 [[package]]
 name = "yansi"
 version = "0.5.1"

Cargo.toml

@@ -26,10 +26,12 @@ 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,6 +92,9 @@ 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
 

clients/client-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "client-core"
-version = "1.0.1"
+version = "1.1.2"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 edition = "2021"
 
@@ -14,11 +14,13 @@ 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"] }
@@ -28,7 +30,6 @@ 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"]}
 
@@ -56,4 +57,5 @@ 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,6 +178,10 @@ 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");
@@ -224,7 +228,9 @@ impl LoopCoverTrafficStream<OsRng> {
                     }
                 }
             }
-            assert!(shutdown.is_shutdown_poll());
+            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+                .await
+                .expect("Task stopped without shutdown called");
             log::debug!("LoopCoverTrafficStream: Exiting");
         })
     }

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

@@ -1,9 +1,9 @@
-use futures::channel::mpsc;
+use client_connections::TransmissionLane;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::anonymous_replies::ReplySurb;
 
-pub type InputMessageSender = mpsc::UnboundedSender<InputMessage>;
-pub type InputMessageReceiver = mpsc::UnboundedReceiver<InputMessage>;
+pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
+pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
 
 #[derive(Debug)]
 pub enum InputMessage {
@@ -11,6 +11,7 @@ pub enum InputMessage {
         recipient: Recipient,
         data: Vec<u8>,
         with_reply_surb: bool,
+        lane: TransmissionLane,
     },
     Reply {
         reply_surb: ReplySurb,
@@ -19,11 +20,17 @@ pub enum InputMessage {
 }
 
 impl InputMessage {
-    pub fn new_fresh(recipient: Recipient, data: Vec<u8>, with_reply_surb: bool) -> Self {
+    pub fn new_fresh(
+        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,6 +69,8 @@ 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");
 
@@ -88,7 +90,9 @@ impl MixTrafficController {
                     }
                 }
             }
-            assert!(shutdown.is_shutdown_poll());
+            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+                .await
+                .expect("Task stopped without shutdown called");
             log::debug!("MixTrafficController: Exiting");
         })
     }

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

@@ -1,5 +1,3 @@
-use std::sync::atomic::AtomicBool;
-
 pub mod cover_traffic_stream;
 pub mod inbound_messages;
 pub mod key_manager;
@@ -9,10 +7,3 @@ 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>) {
-        debug!("Received an ack");
+        trace!("Received an ack");
         let frag_id = match recover_identifier(&self.ack_key, &ack_content)
             .map(FragmentIdentifier::try_from_bytes)
         {
@@ -72,6 +72,8 @@ 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() {
@@ -88,7 +90,9 @@ impl AcknowledgementListener {
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .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,7 +272,9 @@ impl ActionController {
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .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 futures::StreamExt;
+use client_connections::TransmissionLane;
 use log::*;
 use nymsphinx::anonymous_replies::ReplySurb;
 use nymsphinx::preparer::MessagePreparer;
@@ -104,6 +104,7 @@ 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))
@@ -164,37 +165,44 @@ where
     }
 
     async fn on_input_message(&mut self, msg: InputMessage) {
-        let real_messages = match msg {
+        let (real_messages, lane) = 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
-            }
-            InputMessage::Reply { reply_surb, data } => self
-                .handle_reply(reply_surb, data)
-                .await
-                .map(|message| vec![message]),
+                    .map(|message| vec![message]),
+                TransmissionLane::Reply,
+            ),
         };
 
         // 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
-                .unbounded_send(real_messages)
-                .unwrap();
+                .send((real_messages, lane))
+                .await
+                .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.next() => match input_msg {
+                input_msg = self.input_receiver.recv() => match input_msg {
                     Some(input_msg) => {
                         self.on_input_message(input_msg).await;
                     },
@@ -208,14 +216,16 @@ where
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .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.next().await {
+        while let Some(input_msg) = self.input_receiver.recv().await {
             self.on_input_message(input_msg).await;
         }
     }

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

@@ -1,17 +1,21 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use super::action_controller::{Action, ActionSender};
-use super::PendingAcknowledgement;
-use super::RetransmissionRequestReceiver;
+use super::{
+    action_controller::{Action, ActionSender},
+    PendingAcknowledgement, 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::preparer::MessagePreparer;
-use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
+use nymsphinx::{
+    acknowledgements::AckKey, addressing::clients::Recipient, preparer::MessagePreparer,
+};
 use rand::{CryptoRng, Rng};
 use std::sync::{Arc, Weak};
 
@@ -113,15 +117,18 @@ where
 
         // send to `OutQueueControl` to eventually send to the mix network
         self.real_message_sender
-            .unbounded_send(vec![RealMessage::new(
-                prepared_fragment.mix_packet,
-                frag_id,
-            )])
-            .unwrap();
+            .send((
+                vec![RealMessage::new(prepared_fragment.mix_packet, frag_id)],
+                TransmissionLane::Retransmission,
+            ))
+            .await
+            .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() {
@@ -138,7 +145,9 @@ where
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .await
+            .expect("Task stopped without shutdown called");
         log::debug!("RetransmissionRequestListener: Exiting");
     }
 

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

@@ -8,12 +8,15 @@
 use self::{
     acknowledgement_control::AcknowledgementController, real_traffic_stream::OutQueueControl,
 };
-use crate::client::real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors;
-use crate::client::{
-    inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
-    topology_control::TopologyAccessor,
+use crate::{
+    client::{
+        inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
+        real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
+        topology_control::TopologyAccessor,
+    },
+    spawn_future,
 };
-use crate::spawn_future;
+use client_connections::{ConnectionCommandReceiver, LaneQueueLengths};
 use futures::channel::mpsc;
 use gateway_client::AcknowledgementReceiver;
 use log::*;
@@ -103,6 +106,7 @@ 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,
@@ -110,10 +114,12 @@ 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) = mpsc::unbounded();
+        let (real_message_sender, real_message_receiver) = tokio::sync::mpsc::channel(1);
         let (sent_notifier_tx, sent_notifier_rx) = mpsc::unbounded();
 
         let ack_controller_connectors = AcknowledgementControllerConnectors::new(
@@ -159,6 +165,8 @@ 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,7 +4,9 @@
 use crate::client::mix_traffic::BatchMixMessageSender;
 use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender;
 use crate::client::topology_control::TopologyAccessor;
-use futures::channel::mpsc;
+use client_connections::{
+    ConnectionCommand, ConnectionCommandReceiver, ConnectionId, LaneQueueLengths, TransmissionLane,
+};
 use futures::task::{Context, Poll};
 use futures::{Future, Stream, StreamExt};
 use log::*;
@@ -16,7 +18,6 @@ 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;
@@ -27,22 +28,22 @@ 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;
+use self::{
+    sending_delay_controller::SendingDelayController, transmission_buffer::TransmissionBuffer,
+};
+
+mod sending_delay_controller;
+mod transmission_buffer;
+
+#[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()
+}
 
 /// Configurable parameters of the `OutQueueControl`
 pub(crate) struct Config {
@@ -85,101 +86,6 @@ 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,
@@ -203,7 +109,7 @@ where
 
     // To make sure we don't overload the mix_tx channel, we limit the rate we are pushing
     // messages.
-    sending_rate_controller: SendingDelayController,
+    sending_delay_controller: SendingDelayController,
 
     /// Channel used for sending prepared sphinx packets to `MixTrafficController` that sends them
     /// out to the network without any further delays.
@@ -222,10 +128,19 @@ where
     /// Accessor to the common instance of network topology.
     topology_access: TopologyAccessor,
 
-    /// Buffer containing all real messages received. It is first exhausted before more are pulled.
-    received_buffer: VecDeque<RealMessage>,
+    /// Buffer containing all incoming real messages keyed by transmission lane, that we will send
+    /// out to the mixnet.
+    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,
@@ -242,8 +157,9 @@ 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 = mpsc::UnboundedSender<Vec<RealMessage>>;
-type BatchRealMessageReceiver = mpsc::UnboundedReceiver<Vec<RealMessage>>;
+pub(crate) type BatchRealMessageSender =
+    tokio::sync::mpsc::Sender<(Vec<RealMessage>, TransmissionLane)>;
+type BatchRealMessageReceiver = tokio::sync::mpsc::Receiver<(Vec<RealMessage>, TransmissionLane)>;
 
 pub(crate) enum StreamMessage {
     Cover,
@@ -266,22 +182,23 @@ 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_rate_controller: SendingDelayController::new(
-                MIN_DELAY_MULTIPLIER,
-                MAX_DELAY_MULTIPLIER,
-            ),
+            sending_delay_controller: Default::default(),
             mix_tx,
             real_receiver,
             our_full_destination,
             rng,
             topology_access,
-            received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly
+            transmission_buffer: Default::default(),
+            client_connection_rx,
+            lane_queue_lengths,
         }
     }
 
@@ -296,7 +213,7 @@ where
     async fn on_message(&mut self, next_message: StreamMessage) {
         trace!("created new message");
 
-        let next_message = match next_message {
+        let (next_message, fragment_id) = match next_message {
             StreamMessage::Cover => {
                 // TODO for way down the line: in very rare cases (during topology update) we might have
                 // to wait a really tiny bit before actually obtaining the permit hence messing with our
@@ -315,27 +232,41 @@ where
                 }
                 let topology_ref = topology_ref_option.unwrap();
 
-                generate_loop_cover_packet(
-                    &mut self.rng,
-                    topology_ref,
-                    &self.ack_key,
-                    &self.our_full_destination,
-                    self.config.average_ack_delay,
-                    self.config.average_packet_delay,
-                    self.config.cover_packet_size,
+                (
+                    generate_loop_cover_packet(
+                        &mut self.rng,
+                        topology_ref,
+                        &self.ack_key,
+                        &self.our_full_destination,
+                        self.config.average_ack_delay,
+                        self.config.average_packet_delay,
+                        self.config.cover_packet_size,
+                    )
+                    .expect(
+                        "Somehow failed to generate a loop cover message with a valid topology",
+                    ),
+                    None,
                 )
-                .expect("Somehow failed to generate a loop cover message with a valid topology")
             }
             StreamMessage::Real(real_message) => {
-                self.sent_notify(real_message.fragment_id);
-                real_message.mix_packet
+                (real_message.mix_packet, Some(real_message.fragment_id))
             }
         };
 
         if let Err(err) = self.mix_tx.send(vec![next_message]).await {
-            log::error!("Failed to send - channel closed: {}", err);
+            log::error!("Failed to send: {}", err);
         }
 
+        // notify ack controller about sending our message only after we actually managed to push it
+        // through the channel
+        if let Some(fragment_id) = fragment_id {
+            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
@@ -347,44 +278,71 @@ 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_rate_controller.current_multiplier()
+            * self.sending_delay_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_rate_controller.current_multiplier()
+            self.sending_delay_controller.current_multiplier()
         );
 
         // Even just a single used slot is enough to signal backpressure
         if used_slots > 0 {
             log::trace!("Backpressure detected");
-            self.sending_rate_controller.record_backpressure_detected();
+            self.sending_delay_controller.record_backpressure_detected();
         }
 
         // If the buffer is running out, slow down the sending rate
         if self.mix_tx.capacity() == 0
-            && self.sending_rate_controller.not_increased_delay_recently()
+            && self.sending_delay_controller.not_increased_delay_recently()
         {
-            self.sending_rate_controller.increase_delay_multiplier();
+            self.sending_delay_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_rate_controller.is_sending_reliable() {
-            self.sending_rate_controller.decrease_delay_multiplier();
+        if self.sending_delay_controller.is_sending_reliable() {
+            self.sending_delay_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() {
@@ -409,28 +367,32 @@ where
                 next_delay.as_mut().reset(next_poisson_delay);
             }
 
-            // check if we have anything immediately available
-            if let Some(real_available) = self.received_buffer.pop_front() {
-                return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available))));
-            }
-
-            // decide what kind of message to send
-            match Pin::new(&mut self.real_receiver).poll_next(cx) {
+            // On every iteration we get new messages from upstream. Given that these come bunched
+            // in `Vec`, this ensures that on average we will fetch messages faster than we can
+            // send, which is a condition for being able to multiplex sphinx packets from multiple
+            // data streams.
+            match Pin::new(&mut self.real_receiver).poll_recv(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),
 
-                // if there are more messages available, return first one and store the rest
-                Poll::Ready(Some(real_messages)) => {
-                    self.received_buffer = real_messages.into();
-                    // we MUST HAVE received at least ONE message
-                    Poll::Ready(Some(StreamMessage::Real(Box::new(
-                        self.received_buffer.pop_front().unwrap(),
-                    ))))
+                Poll::Ready(Some((real_messages, conn_id))) => {
+                    log::trace!("handling real_messages: size: {}", real_messages.len());
+
+                    self.transmission_buffer.store(&conn_id, real_messages);
+                    let real_next = self.pop_next_message().expect("Just stored one");
+
+                    Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
                 }
 
-                // otherwise construct a dummy one
-                Poll::Pending => Poll::Ready(Some(StreamMessage::Cover)),
+                Poll::Pending => {
+                    if let Some(real_next) = self.pop_next_message() {
+                        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
@@ -452,32 +414,36 @@ where
     }
 
     fn poll_immediate(&mut self, cx: &mut Context<'_>) -> Poll<Option<StreamMessage>> {
-        // check if we have anything immediately available
-        if let Some(real_available) = self.received_buffer.pop_front() {
-            // if there are more messages immediately available, notify the runtime
-            // because we should be polled again
-            if !self.received_buffer.is_empty() {
-                cx.waker().wake_by_ref()
+        // Start by checking if we have any incoming messages about closed connections
+        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!(),
             }
-            return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available))));
         }
 
-        match Pin::new(&mut self.real_receiver).poll_next(cx) {
+        match Pin::new(&mut self.real_receiver).poll_recv(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),
 
-            // if there are more messages available, return first one and store the rest
-            Poll::Ready(Some(real_messages)) => {
-                self.received_buffer = real_messages.into();
-                // we MUST HAVE received at least ONE message
-                Poll::Ready(Some(StreamMessage::Real(Box::new(
-                    self.received_buffer.pop_front().unwrap(),
-                ))))
+            Poll::Ready(Some((real_messages, conn_id))) => {
+                log::trace!("handling real_messages: size: {}", real_messages.len());
+
+                // First store what we got for the given connection id
+                self.transmission_buffer.store(&conn_id, real_messages);
+                let real_next = self.pop_next_message().expect("we just added one");
+
+                Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
             }
 
-            // if there's nothing, then there's nothing
-            Poll::Pending => Poll::Pending,
+            Poll::Pending => {
+                if let Some(real_next) = self.pop_next_message() {
+                    Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
+                } else {
+                    Poll::Pending
+                }
+            }
         }
     }
 
@@ -492,28 +458,73 @@ 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");
                 }
-                next_message = self.next() => match next_message {
-                    Some(next_message) => {
-                        self.on_message(next_message).await;
-                    },
-                    None => {
-                        log::trace!("OutQueueControl: Stopping since channel closed");
-                        break;
-                    }
+                _ = status_timer.tick() => {
+                    self.log_status();
+                }
+                _ = infrequent_status_timer.tick() => {
+                    self.log_status_infrequent();
+                }
+                next_message = self.next() => if let Some(next_message) = next_message {
+                    self.on_message(next_message).await;
+                } else {
+                    log::trace!("OutQueueControl: Stopping since channel closed");
+                    break;
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .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

@@ -0,0 +1,124 @@
+// 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

@@ -0,0 +1,211 @@
+// 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>>) {
-        debug!(
+        trace!(
             "Processing {:?} new message that might get added to the buffer!",
             msgs.len()
         );
@@ -322,6 +322,8 @@ 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! {
@@ -340,7 +342,9 @@ impl RequestReceiver {
                 },
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .await
+            .expect("Task stopped without shutdown called");
         log::debug!("RequestReceiver: Exiting");
     }
 
@@ -372,6 +376,8 @@ 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! {
@@ -389,7 +395,9 @@ impl FragmentedMessageReceiver {
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
+        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            .await
+            .expect("Task stopped without shutdown called");
         log::debug!("FragmentedMessageReceiver: Exiting");
     }
 

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

@@ -8,10 +8,13 @@ use nymsphinx::anonymous_replies::{
 };
 use std::path::Path;
 
-#[derive(Debug)]
+#[derive(Debug, thiserror::Error)]
 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

@@ -57,24 +57,15 @@ impl<'a> TopologyReadPermit<'a> {
     ) -> Option<&'a NymTopology> {
         // Note: implicit deref with Deref for TopologyReadPermit is happening here
         let topology_ref_option = self.permit.as_ref();
-        match topology_ref_option {
-            None => None,
-            Some(topology_ref) => {
-                // see if it's possible to route the packet to both gateways
-                if !topology_ref.can_construct_path_through(DEFAULT_NUM_MIX_HOPS)
-                    || !topology_ref.gateway_exists(ack_recipient.gateway())
-                    || if let Some(packet_recipient) = packet_recipient {
-                        !topology_ref.gateway_exists(packet_recipient.gateway())
-                    } else {
-                        false
-                    }
-                {
-                    None
+        topology_ref_option.as_ref().filter(|topology_ref| {
+            !(!topology_ref.can_construct_path_through(DEFAULT_NUM_MIX_HOPS)
+                || !topology_ref.gateway_exists(ack_recipient.gateway())
+                || if let Some(packet_recipient) = packet_recipient {
+                    !topology_ref.gateway_exists(packet_recipient.gateway())
                 } else {
-                    Some(topology_ref)
-                }
-            }
-        }
+                    false
+                })
+        })
     }
 }
 
@@ -147,7 +138,7 @@ impl TopologyRefresherConfig {
 }
 
 pub struct TopologyRefresher {
-    validator_client: validator_client::ApiClient,
+    validator_client: validator_client::client::ApiClient,
     client_version: String,
 
     validator_api_urls: Vec<Url>,
@@ -163,7 +154,9 @@ impl TopologyRefresher {
         cfg.validator_api_urls.shuffle(&mut thread_rng());
 
         TopologyRefresher {
-            validator_client: validator_client::ApiClient::new(cfg.validator_api_urls[0].clone()),
+            validator_client: validator_client::client::ApiClient::new(
+                cfg.validator_api_urls[0].clone(),
+            ),
             client_version: cfg.client_version,
             validator_api_urls: cfg.validator_api_urls,
             topology_accessor,
@@ -193,13 +186,10 @@ impl TopologyRefresher {
     /// # Arguments
     ///
     /// * `topology`: active topology constructed from validator api data
-    /// * `mixnodes_count`: total number of active mixnodes
-    fn check_layer_distribution(
-        &self,
-        active_topology: &NymTopology,
-        mixnodes_count: usize,
-    ) -> bool {
+    fn check_layer_distribution(&self, active_topology: &NymTopology) -> bool {
         let mixes = active_topology.mixes();
+        let mixnodes_count = active_topology.num_mixnodes();
+
         if active_topology.gateways().is_empty() {
             return false;
         }
@@ -264,11 +254,10 @@ impl TopologyRefresher {
             Ok(gateways) => gateways,
         };
 
-        let mixnodes_count = mixnodes.len();
         let topology = nym_topology_from_detailed(mixnodes, gateways)
             .filter_system_version(&self.client_version);
 
-        if !self.check_layer_distribution(&topology, mixnodes_count) {
+        if !self.check_layer_distribution(&topology) {
             warn!("The current filtered active topology has extremely skewed layer distribution. It cannot be used.");
             None
         } else {
@@ -318,7 +307,9 @@ impl TopologyRefresher {
                     },
                 }
             }
-            assert!(shutdown.is_shutdown_poll());
+            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+                .await
+                .expect("Task stopped without shutdown called");
             log::debug!("TopologyRefresher: Exiting");
         })
     }

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

@@ -1,7 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use config::NymConfig;
+use config::{NymConfig, DB_FILE_NAME};
 use nymsphinx::params::PacketSize;
 use serde::{Deserialize, Serialize};
 use std::marker::PhantomData;
@@ -125,6 +125,10 @@ 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;
     }
@@ -179,6 +183,10 @@ 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()
     }
@@ -306,6 +314,10 @@ 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>,
 
@@ -354,6 +366,7 @@ 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(),
@@ -399,7 +412,7 @@ impl<T: NymConfig> Client<T> {
         T::default_data_directory(Some(id)).join("reply_key_store")
     }
     fn default_database_path(id: &str) -> PathBuf {
-        T::default_data_directory(Some(id)).join("db.sqlite")
+        T::default_data_directory(Some(id)).join(DB_FILE_NAME)
     }
 }
 

clients/client-core/src/error.rs

@@ -24,4 +24,9 @@ pub enum ClientCoreError {
     ListOfValidatorApisIsEmpty,
     #[error("Could not load existing gateway configuration: {0}")]
     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::ApiClient::new(validator_api.clone());
+    let validator_client = validator_client::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/Cargo.toml

@@ -6,11 +6,9 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-async-trait = "0.1.52"
 bip39 = "1.0.1"
 cfg-if = "0.1"
 clap = { version = "3.2", features = ["cargo", "derive"] }
-pickledb = "0.4.1"
 rand = "0.7.3"
 serde = { version = "1.0", features = ["derive"] }
 thiserror = "1.0"

clients/credential/src/commands.rs

@@ -1,34 +1,29 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use async_trait::async_trait;
 use clap::{Args, Subcommand};
 use completions::ArgShell;
-use pickledb::PickleDb;
 use rand::rngs::OsRng;
 use std::str::FromStr;
 
-use coconut_interface::{Attribute, Base58, BlindSignRequest, Bytable, Parameters};
+use coconut_interface::{Base58, Parameters};
 use credential_storage::storage::Storage;
 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::VOUCHER_INFO;
+use network_defaults::{NymNetworkDetails, VOUCHER_INFO};
 use validator_client::nymd::tx::Hash;
+use validator_client::{CoconutApiClient, Config};
 
 use crate::client::Client;
 use crate::error::{CredentialClientError, Result};
-use crate::state::{KeyPair, RequestData, State};
+use crate::state::{KeyPair, State};
 
 #[derive(Subcommand)]
-pub(crate) enum Commands {
-    /// Deposit funds for buying coconut credential
-    Deposit(Deposit),
-    /// Lists the tx hashes of previous deposits
-    ListDeposits(ListDeposits),
-    /// Get a credential for a given deposit
-    GetCredential(GetCredential),
+pub(crate) enum Command {
+    /// Run the binary
+    Run(Run),
 
     /// Generate shell completions
     Completions(ArgShell),
@@ -37,163 +32,81 @@ pub(crate) enum Commands {
     GenerateFigSpec,
 }
 
-#[async_trait]
-pub(crate) trait Execute {
-    async fn execute(&self, db: &mut PickleDb, shared_storage: PersistentStorage) -> Result<()>;
-}
+#[derive(Args)]
+pub(crate) struct Run {
+    /// Home directory of the client that is supposed to use the credential.
+    #[clap(long)]
+    pub(crate) client_home_directory: std::path::PathBuf,
 
-#[derive(Args, Clone)]
-pub(crate) struct Deposit {
     /// The nymd URL that should be used
     #[clap(long)]
-    nymd_url: String,
-    /// A mnemonic for the account that does the deposit
+    pub(crate) nymd_url: String,
+
+    /// A mnemonic for the account that buys the credential
     #[clap(long)]
-    mnemonic: String,
-    /// The amount that needs to be deposited
+    pub(crate) mnemonic: String,
+
+    /// The amount of utokens the credential will hold
     #[clap(long)]
-    amount: u64,
+    pub(crate) amount: u64,
 }
 
-#[async_trait]
-impl Execute for Deposit {
-    async fn execute(&self, db: &mut PickleDb, _shared_storage: PersistentStorage) -> Result<()> {
-        let mut rng = OsRng;
-        let signing_keypair = KeyPair::from(identity::KeyPair::new(&mut rng));
-        let encryption_keypair = KeyPair::from(encryption::KeyPair::new(&mut rng));
+pub(crate) async fn deposit(nymd_url: &str, mnemonic: &str, amount: u64) -> Result<State> {
+    let mut rng = OsRng;
+    let signing_keypair = KeyPair::from(identity::KeyPair::new(&mut rng));
+    let encryption_keypair = KeyPair::from(encryption::KeyPair::new(&mut rng));
 
-        let client = Client::new(&self.nymd_url, &self.mnemonic);
-        let tx_hash = client
-            .deposit(
-                self.amount,
-                signing_keypair.public_key.clone(),
-                encryption_keypair.public_key.clone(),
-                None,
-            )
-            .await?;
+    let client = Client::new(nymd_url, mnemonic);
+    let tx_hash = client
+        .deposit(
+            amount,
+            signing_keypair.public_key.clone(),
+            encryption_keypair.public_key.clone(),
+            None,
+        )
+        .await?;
 
-        let state = State {
-            amount: self.amount,
-            tx_hash: tx_hash.clone(),
-            signing_keypair,
-            encryption_keypair,
-            blind_request_data: None,
-            signature: None,
-        };
-        db.set(&tx_hash, &state).unwrap();
+    let state = State {
+        amount,
+        tx_hash,
+        signing_keypair,
+        encryption_keypair,
+    };
 
-        println!("{:?}", state);
-
-        Ok(())
-    }
+    Ok(state)
 }
 
-#[derive(Args, Clone)]
-pub(crate) struct ListDeposits {}
+pub(crate) async fn get_credential(state: &State, shared_storage: PersistentStorage) -> Result<()> {
+    let network_details = NymNetworkDetails::new_from_env();
+    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?;
 
-#[async_trait]
-impl Execute for ListDeposits {
-    async fn execute(&self, db: &mut PickleDb, _shared_storage: PersistentStorage) -> Result<()> {
-        for kv in db.iter() {
-            println!("{:?}", kv.get_value::<State>());
-        }
+    let params = Parameters::new(TOTAL_ATTRIBUTES).unwrap();
+    let bandwidth_credential_attributes = BandwidthVoucher::new(
+        &params,
+        state.amount.to_string(),
+        VOUCHER_INFO.to_string(),
+        Hash::from_str(&state.tx_hash).map_err(|_| CredentialClientError::InvalidTxHash)?,
+        identity::PrivateKey::from_base58_string(&state.signing_keypair.private_key)?,
+        encryption::PrivateKey::from_base58_string(&state.encryption_keypair.private_key)?,
+    );
 
-        Ok(())
-    }
-}
-
-#[derive(Args, Clone)]
-pub(crate) struct GetCredential {
-    /// The hash of a successful deposit transaction
-    #[clap(long)]
-    tx_hash: String,
-    /// The URLs to the validator-api endpoints the are run as coconut signer authorities, separated
-    /// by comma (,)
-    #[clap(long)]
-    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))]
-    __no_request: bool,
-}
-
-#[async_trait]
-impl Execute for GetCredential {
-    async fn execute(&self, db: &mut PickleDb, shared_storage: PersistentStorage) -> Result<()> {
-        let mut state = db
-            .get::<State>(&self.tx_hash)
-            .ok_or(CredentialClientError::NoDeposit)?;
-
-        let urls = config::parse_validators(&self.signer_authorities);
-
-        let params = Parameters::new(TOTAL_ATTRIBUTES).unwrap();
-        let bandwidth_credential_attributes = if self.__no_request {
-            if let Some(blind_request_data) = state.blind_request_data {
-                let serial_number =
-                    Attribute::try_from_byte_slice(&blind_request_data.serial_number)
-                        .map_err(|_| CredentialClientError::CorruptedBlindSignRequest)?;
-                let binding_number =
-                    Attribute::try_from_byte_slice(&blind_request_data.binding_number)
-                        .map_err(|_| CredentialClientError::CorruptedBlindSignRequest)?;
-                let pedersen_commitments_openings = vec![
-                    Attribute::try_from_byte_slice(&blind_request_data.first_attribute)
-                        .map_err(|_| CredentialClientError::CorruptedBlindSignRequest)?,
-                    Attribute::try_from_byte_slice(&blind_request_data.second_attribute)
-                        .map_err(|_| CredentialClientError::CorruptedBlindSignRequest)?,
-                ];
-                let blind_sign_request =
-                    BlindSignRequest::from_bytes(blind_request_data.blind_sign_req.as_slice())
-                        .map_err(|_| CredentialClientError::CorruptedBlindSignRequest)?;
-                BandwidthVoucher::new_with_blind_sign_req(
-                    [serial_number, binding_number],
-                    [&state.amount.to_string(), VOUCHER_INFO],
-                    Hash::from_str(&self.tx_hash)
-                        .map_err(|_| CredentialClientError::InvalidTxHash)?,
-                    identity::PrivateKey::from_base58_string(&state.signing_keypair.private_key)?,
-                    encryption::PrivateKey::from_base58_string(
-                        &state.encryption_keypair.private_key,
-                    )?,
-                    pedersen_commitments_openings,
-                    blind_sign_request,
-                )
-            } else {
-                return Err(CredentialClientError::NoLocalBlindSignRequest);
-            }
-        } else {
-            BandwidthVoucher::new(
-                &params,
-                state.amount.to_string(),
-                VOUCHER_INFO.to_string(),
-                Hash::from_str(&self.tx_hash).map_err(|_| CredentialClientError::InvalidTxHash)?,
-                identity::PrivateKey::from_base58_string(&state.signing_keypair.private_key)?,
-                encryption::PrivateKey::from_base58_string(&state.encryption_keypair.private_key)?,
-            )
-        };
-
-        // Back up the blind sign req data, in case of sporadic failures
-        state.blind_request_data = Some(RequestData::new(
-            bandwidth_credential_attributes.get_private_attributes(),
-            bandwidth_credential_attributes.pedersen_commitments_openings(),
-            bandwidth_credential_attributes.blind_sign_request(),
-        )?);
-        db.set(&self.tx_hash, &state).unwrap();
-
-        let signature =
-            obtain_aggregate_signature(&params, &bandwidth_credential_attributes, &urls).await?;
-        shared_storage
-            .insert_coconut_credential(
-                state.amount.to_string(),
-                VOUCHER_INFO.to_string(),
-                bandwidth_credential_attributes.get_private_attributes()[0].to_bs58(),
-                bandwidth_credential_attributes.get_private_attributes()[1].to_bs58(),
-                signature.to_bs58(),
-            )
-            .await?;
-        state.signature = Some(signature.to_bs58());
-        db.set(&self.tx_hash, &state).unwrap();
-
-        println!("Signature: {:?}", state.signature);
-
-        Ok(())
-    }
+    let signature = obtain_aggregate_signature(
+        &params,
+        &bandwidth_credential_attributes,
+        &coconut_api_clients,
+    )
+    .await?;
+    shared_storage
+        .insert_coconut_credential(
+            state.amount.to_string(),
+            VOUCHER_INFO.to_string(),
+            bandwidth_credential_attributes.get_private_attributes()[0].to_bs58(),
+            bandwidth_credential_attributes.get_private_attributes()[1].to_bs58(),
+            signature.to_bs58(),
+        )
+        .await?;
+
+    Ok(())
 }

clients/credential/src/error.rs

@@ -8,6 +8,7 @@ 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>;
 
@@ -16,21 +17,12 @@ 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),
 
-    #[error("No previous deposit with that tx hash")]
-    NoDeposit,
-
-    #[error("Wrong number of attributes")]
-    WrongAttributeNumber,
-
-    #[error("Could not find any backed up blind sign request data")]
-    NoLocalBlindSignRequest,
-
-    #[error("The local blind sign request data is corrupted")]
-    CorruptedBlindSignRequest,
-
     #[error("The tx hash provided is not valid")]
     InvalidTxHash,
 

clients/credential/src/main.rs

@@ -9,59 +9,41 @@ cfg_if::cfg_if! {
         mod error;
         mod state;
 
-        use commands::{Commands, Execute};
         use error::Result;
         use network_defaults::setup_env;
-        use clap::CommandFactory;
         use completions::fig_generate;
+        use commands::*;
+        use config::{DATA_DIR, DB_FILE_NAME};
 
-        use clap::Parser;
-        use pickledb::{PickleDb, PickleDbDumpPolicy, SerializationMethod};
+        use clap::{CommandFactory, Parser};
 
         #[derive(Parser)]
         #[clap(author = "Nymtech", version, about)]
         struct Cli {
             /// Path pointing to an env file that configures the client.
-            #[clap(long)]
+            #[clap(short, long)]
             pub(crate) config_env_file: Option<std::path::PathBuf>,
 
-            /// Path where the sqlite credental database will be located.
-            /// It should point to a $HOME/$CLIENT_ID/data/db.sqlite file of
-            /// the client that is supposed to use the credential.
-            #[clap(long)]
-            pub(crate) credential_db_path: std::path::PathBuf,
-
             #[clap(subcommand)]
-            command: Commands,
+            pub(crate) command: Command,
         }
 
         #[tokio::main]
         async fn main() -> Result<()> {
             let args = Cli::parse();
             setup_env(args.config_env_file.clone());
-
-            let shared_storage = credential_storage::initialise_storage(args.credential_db_path.clone()).await;
-            let mut db = match PickleDb::load(
-                "credential.db",
-                PickleDbDumpPolicy::AutoDump,
-                SerializationMethod::Json,
-            ) {
-                Ok(db) => db,
-                Err(_) => PickleDb::new(
-                    "credential.db",
-                    PickleDbDumpPolicy::AutoDump,
-                    SerializationMethod::Json,
-                ),
-            };
-
             let bin_name = "nym-credential-client";
 
-            match &args.command {
-                Commands::Deposit(m) => m.execute(&mut db, shared_storage).await?,
-                Commands::ListDeposits(m) => m.execute(&mut db, shared_storage).await?,
-                Commands::GetCredential(m) => m.execute(&mut db, shared_storage).await?,
-                Commands::Completions(s) => s.generate(&mut crate::Cli::into_app(), bin_name),
-                Commands::GenerateFigSpec => fig_generate(&mut crate::Cli::into_app(), bin_name)
+            match args.command {
+                Command::Run(r) => {
+                    let db_path = r.client_home_directory.join(DATA_DIR).join(DB_FILE_NAME);
+                    let shared_storage = credential_storage::initialise_storage(db_path).await;
+
+                    let state = deposit(&r.nymd_url, &r.mnemonic, r.amount).await?;
+                    get_credential(&state, shared_storage).await?;
+                }
+                Command::Completions(c) => c.generate(&mut crate::Cli::into_app(), bin_name),
+                Command::GenerateFigSpec => fig_generate(&mut crate::Cli::into_app(), bin_name)
             }
 
             Ok(())

clients/credential/src/state.rs

@@ -1,13 +1,10 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use coconut_interface::{Attribute, BlindSignRequest, Bytable, PrivateAttribute};
 use serde::{Deserialize, Serialize};
 
 use crypto::asymmetric::{encryption, identity};
 
-use crate::error::{CredentialClientError, Result};
-
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub(crate) struct KeyPair {
     pub public_key: String,
@@ -38,35 +35,4 @@ pub(crate) struct State {
     pub tx_hash: String,
     pub signing_keypair: KeyPair,
     pub encryption_keypair: KeyPair,
-    pub blind_request_data: Option<RequestData>,
-    pub signature: Option<String>,
-}
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub(crate) struct RequestData {
-    pub serial_number: Vec<u8>,
-    pub binding_number: Vec<u8>,
-    pub first_attribute: Vec<u8>,
-    pub second_attribute: Vec<u8>,
-    pub blind_sign_req: Vec<u8>,
-}
-
-impl RequestData {
-    pub fn new(
-        private_attributes: Vec<PrivateAttribute>,
-        attributes: &[Attribute],
-        blind_sign_request: &BlindSignRequest,
-    ) -> Result<Self> {
-        if private_attributes.len() != 2 || attributes.len() != 2 {
-            Err(CredentialClientError::WrongAttributeNumber)
-        } else {
-            Ok(RequestData {
-                serial_number: private_attributes[0].to_byte_vec(),
-                binding_number: private_attributes[1].to_byte_vec(),
-                first_attribute: attributes[0].to_byte_vec(),
-                second_attribute: attributes[1].to_byte_vec(),
-                blind_sign_req: blind_sign_request.to_bytes(),
-            })
-        }
-    }
 }

clients/native/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
 name = "nym-client"
-version = "1.0.2"
+version = "1.1.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.56"
+rust-version = "1.65"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -27,11 +27,13 @@ pretty_env_logger = "0.4" # for formatting log messages
 rand = { version = "0.7.3", features = ["wasm-bindgen"] } # rng-related traits + some rng implementation to use
 serde = { version = "1.0.104", features = ["derive"] } # for config serialization/deserialization
 sled = "0.34" # for storage of replySURB decryption keys
+thiserror = "1.0.34"
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal"] } # async runtime
 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" }
@@ -49,6 +51,7 @@ 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,6 +43,7 @@ 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...");
@@ -91,6 +92,7 @@ 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,6 +23,13 @@ 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,6 +1,9 @@
 // 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,
@@ -18,6 +21,7 @@ use client_core::client::topology_control::{
     TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
 };
 use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
+use client_core::error::ClientCoreError;
 use crypto::asymmetric::identity;
 use futures::channel::mpsc;
 use gateway_client::bandwidth::BandwidthController;
@@ -30,9 +34,11 @@ 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};
+use crate::error::ClientError;
 use crate::websocket;
 
 pub(crate) mod config;
@@ -108,6 +114,7 @@ impl NymClient {
         stream.start_with_shutdown(shutdown);
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn start_real_traffic_controller(
         &self,
         topology_accessor: TopologyAccessor,
@@ -115,6 +122,8 @@ 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(
@@ -144,6 +153,8 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
+            lane_queue_lengths,
+            client_connection_rx,
         )
         .start_with_shutdown(shutdown);
     }
@@ -190,11 +201,37 @@ impl NymClient {
             .expect("provided gateway id is invalid!");
 
         #[cfg(feature = "coconut")]
-        let bandwidth_controller = BandwidthController::new(
-            credential_storage::initialise_storage(self.config.get_base().get_database_path())
-                .await,
-            self.config.get_base().get_validator_api_endpoints(),
-        );
+        let bandwidth_controller = {
+            let details = network_defaults::NymNetworkDetails::new_from_env();
+            let mut client_config =
+                validator_client::Config::try_from_nym_network_details(&details)
+                    .expect("failed to construct validator client config");
+            let nymd_url = self
+                .config
+                .get_base()
+                .get_validator_endpoints()
+                .pop()
+                .expect("No nymd validator endpoint provided");
+            let api_url = self
+                .config
+                .get_base()
+                .get_validator_api_endpoints()
+                .pop()
+                .expect("No validator api endpoint provided");
+            // overwrite env configuration with config URLs
+            client_config = client_config.with_urls(nymd_url, api_url);
+            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())
@@ -232,7 +269,7 @@ impl NymClient {
         &mut self,
         topology_accessor: TopologyAccessor,
         shutdown: ShutdownListener,
-    ) {
+    ) -> Result<(), ClientError> {
         let topology_refresher_config = TopologyRefresherConfig::new(
             self.config.get_base().get_validator_api_endpoints(),
             self.config.get_base().get_topology_refresh_rate(),
@@ -247,14 +284,16 @@ impl NymClient {
 
         // TODO: a slightly more graceful termination here
         if !topology_refresher.is_topology_routable().await {
-            panic!(
-                "The current network topology seem to be insufficient to route any packets through\
+            log::error!(
+                "The current network topology seem to be insufficient to route any packets through \
                 - check if enough nodes and a gateway are online"
             );
+            return Err(ClientCoreError::InsufficientNetworkTopology.into());
         }
 
         info!("Starting topology refresher...");
         topology_refresher.start_with_shutdown(shutdown);
+        Ok(())
     }
 
     // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
@@ -275,11 +314,18 @@ 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(msg_input, buffer_requester, self.as_mix_recipient());
+        let websocket_handler = websocket::Handler::new(
+            msg_input,
+            client_connection_tx,
+            buffer_requester,
+            &self.as_mix_recipient(),
+            shared_lane_queue_lengths,
+        );
 
         websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler);
     }
@@ -287,27 +333,35 @@ 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 fn send_message(&mut self, recipient: Recipient, message: Vec<u8>, with_reply_surb: bool) {
-        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb);
+    pub async fn send_message(
+        &mut self,
+        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!")
-            .unbounded_send(input_msg)
-            .unwrap();
+            .send(input_msg)
+            .await
+            .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 fn send_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) {
+    pub async 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!")
-            .unbounded_send(input_msg)
-            .unwrap();
+            .send(input_msg)
+            .await
+            .expect("InputMessageReceiver has stopped receiving!");
     }
 
     /// EXPERIMENTAL DIRECT RUST API
@@ -328,8 +382,8 @@ impl NymClient {
     }
 
     /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
-    pub async fn run_forever(&mut self) {
-        let shutdown = self.start().await;
+    pub async fn run_forever(&mut self) -> Result<(), ClientError> {
+        let shutdown = self.start().await?;
         wait_for_signal().await;
 
         println!(
@@ -346,9 +400,10 @@ impl NymClient {
         //shutdown.wait_for_shutdown().await;
 
         log::info!("Stopping nym-client");
+        Ok(())
     }
 
-    pub async fn start(&mut self) -> ShutdownNotifier {
+    pub async fn start(&mut self) -> Result<ShutdownNotifier, ClientError> {
         info!("Starting nym client");
         // channels for inter-component communication
         // TODO: make the channels be internally created by the relevant components
@@ -363,7 +418,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) = mpsc::unbounded::<InputMessage>();
+        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
 
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
@@ -371,7 +426,10 @@ impl NymClient {
 
         let reply_key_storage =
             ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
-                .expect("Failed to load reply key storage!");
+                .tap_err(|err| {
+                    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();
@@ -379,7 +437,7 @@ impl NymClient {
         // 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(), shutdown.subscribe())
-            .await;
+            .await?;
         self.start_received_messages_buffer_controller(
             received_buffer_request_receiver,
             mixnet_messages_receiver,
@@ -398,12 +456,22 @@ 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(),
         );
 
@@ -420,9 +488,12 @@ impl NymClient {
         }
 
         match self.config.get_socket_type() {
-            SocketType::WebSocket => {
-                self.start_websocket_listener(received_buffer_request_sender, input_sender)
-            }
+            SocketType::WebSocket => self.start_websocket_listener(
+                received_buffer_request_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
@@ -443,6 +514,6 @@ impl NymClient {
         info!("Client startup finished!");
         info!("The address of this client is: {}", self.as_mix_recipient());
 
-        shutdown
+        Ok(shutdown)
     }
 }

clients/native/src/commands/init.rs

@@ -25,9 +25,13 @@ pub(crate) struct Init {
     #[clap(long)]
     force_register_gateway: bool,
 
-    /// Comma separated list of rest endpoints of the validators
+    /// Comma separated list of rest endpoints of the nymd validators
     #[clap(long)]
-    validators: Option<String>,
+    nymd_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)]
@@ -52,7 +56,8 @@ pub(crate) struct Init {
 impl From<Init> for OverrideConfig {
     fn from(init_config: Init) -> Self {
         OverrideConfig {
-            validators: init_config.validators,
+            nymd_validators: init_config.nymd_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

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::client::config::{Config, SocketType};
+use crate::error::ClientError;
 use clap::CommandFactory;
 use clap::{Parser, Subcommand};
 use completions::{fig_generate, ArgShell};
@@ -49,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(long)]
+    #[clap(short, long)]
     pub(crate) config_env_file: Option<std::path::PathBuf>,
 
     #[clap(subcommand)]
@@ -74,7 +75,8 @@ pub(crate) enum Commands {
 
 // Configuration that can be overridden.
 pub(crate) struct OverrideConfig {
-    validators: Option<String>,
+    nymd_validators: Option<String>,
+    api_validators: Option<String>,
     disable_socket: bool,
     port: Option<u16>,
     fastmode: bool,
@@ -83,20 +85,32 @@ pub(crate) struct OverrideConfig {
     enabled_credentials_mode: bool,
 }
 
-pub(crate) async fn execute(args: &Cli) {
+pub(crate) async fn execute(args: &Cli) -> Result<(), ClientError> {
     let bin_name = "nym-native-client";
 
     match &args.command {
         Commands::Init(m) => init::execute(m).await,
-        Commands::Run(m) => run::execute(m).await,
+        Commands::Run(m) => run::execute(m).await?,
         Commands::Upgrade(m) => upgrade::execute(m),
         Commands::Completions(s) => s.generate(&mut Cli::into_app(), bin_name),
         Commands::GenerateFigSpec => fig_generate(&mut Cli::into_app(), bin_name),
     }
+    Ok(())
 }
 
 pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Config {
-    if let Some(raw_validators) = args.validators {
+    if let Some(raw_validators) = args.nymd_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

@@ -4,6 +4,7 @@
 use crate::{
     client::{config::Config, NymClient},
     commands::{override_config, OverrideConfig},
+    error::ClientError,
 };
 
 use clap::Args;
@@ -17,9 +18,13 @@ pub(crate) struct Run {
     #[clap(long)]
     id: String,
 
-    /// Comma separated list of rest endpoints of the validators
+    /// Comma separated list of rest endpoints of the nymd validators
     #[clap(long)]
-    validators: Option<String>,
+    nymd_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
@@ -44,7 +49,8 @@ pub(crate) struct Run {
 impl From<Run> for OverrideConfig {
     fn from(run_config: Run) -> Self {
         OverrideConfig {
-            validators: run_config.validators,
+            nymd_validators: run_config.nymd_validators,
+            api_validators: run_config.api_validators,
             disable_socket: run_config.disable_socket,
             port: run_config.port,
             fastmode: false,
@@ -73,14 +79,14 @@ fn version_check(cfg: &Config) -> bool {
     }
 }
 
-pub(crate) async fn execute(args: &Run) {
+pub(crate) async fn execute(args: &Run) -> Result<(), ClientError> {
     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;
+            return Err(ClientError::FailedToLoadConfig(id.to_string()));
         }
     };
 
@@ -89,8 +95,8 @@ pub(crate) async fn execute(args: &Run) {
 
     if !version_check(&config) {
         error!("failed the local version check");
-        return;
+        return Err(ClientError::FailedLocalVersionCheck);
     }
 
-    NymClient::new(config).run_forever().await;
+    NymClient::new(config).run_forever().await
 }

clients/native/src/error.rs

@@ -0,0 +1,25 @@
+use client_core::{client::reply_key_storage::ReplyKeyStorageError, error::ClientCoreError};
+use crypto::asymmetric::identity::Ed25519RecoveryError;
+use gateway_client::error::GatewayClientError;
+use validator_client::ValidatorClientError;
+
+#[derive(thiserror::Error, Debug)]
+pub enum ClientError {
+    #[error("I/O error: {0}")]
+    IoError(#[from] std::io::Error),
+    #[error("Gateway client error: {0}")]
+    GatewayClientError(#[from] GatewayClientError),
+    #[error("Ed25519 error: {0}")]
+    Ed25519RecoveryError(#[from] Ed25519RecoveryError),
+    #[error("Validator client error: {0}")]
+    ValidatorClientError(#[from] ValidatorClientError),
+    #[error("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),
+    #[error("Failed local version check, client and config mismatch")]
+    FailedLocalVersionCheck,
+}

clients/native/src/lib.rs

@@ -2,4 +2,5 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod client;
+pub mod error;
 pub mod websocket;

clients/native/src/main.rs

@@ -2,21 +2,23 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::{crate_version, Parser};
+use error::ClientError;
 use logging::setup_logging;
 use network_defaults::setup_env;
 
 pub mod client;
 pub mod commands;
+pub mod error;
 pub mod websocket;
 
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<(), ClientError> {
     setup_logging();
     println!("{}", banner());
 
     let args = commands::Cli::parse();
     setup_env(args.config_env_file.clone());
-    commands::execute(&args).await;
+    commands::execute(&args).await
 }
 
 fn banner() -> String {

clients/native/src/websocket/handler.rs

@@ -1,6 +1,9 @@
 // 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::{
@@ -34,10 +37,12 @@ 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`
@@ -45,10 +50,12 @@ 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(),
         }
     }
 }
@@ -64,38 +71,85 @@ 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,
         }
     }
 
-    fn handle_send(
+    async fn handle_send(
         &mut self,
-        recipient: Recipient,
+        recipient: &Recipient,
         message: Vec<u8>,
         with_reply_surb: bool,
+        connection_id: Option<u64>,
     ) -> Option<ServerResponse> {
-        // the ack control is now responsible for chunking, etc.
-        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb);
-        self.msg_input.unbounded_send(input_msg).unwrap();
+        // 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)
+        });
 
-        None
+        // the ack control is now responsible for chunking, etc.
+        let input_msg = InputMessage::new_fresh(*recipient, message, with_reply_surb, lane);
+        self.msg_input
+            .send(input_msg)
+            .await
+            .expect("InputMessageReceiver has stopped receiving!");
+
+        // Only reply back with a `LaneQueueLength` if the sender providided a connection id
+        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))
     }
 
-    fn handle_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) -> Option<ServerResponse> {
+    async fn handle_reply(
+        &mut self,
+        reply_surb: ReplySurb,
+        message: Vec<u8>,
+    ) -> Option<ServerResponse> {
         if message.len() > ReplySurb::max_msg_len(Default::default()) {
-            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()))));
+            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()))
+                    )
+                );
         }
 
         let input_msg = InputMessage::new_reply(reply_surb, message);
-        self.msg_input.unbounded_send(input_msg).unwrap();
+        self.msg_input
+            .send(input_msg)
+            .await
+            .expect("InputMessageReceiver has stopped receiving!");
 
         None
     }
@@ -104,22 +158,48 @@ impl Handler {
         ServerResponse::SelfAddress(self.self_full_address)
     }
 
-    fn handle_request(&mut self, request: ClientRequest) -> Option<ServerResponse> {
+    fn handle_closed_connection(&self, connection_id: u64) -> 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,
-            } => self.handle_send(recipient, message, with_reply_surb),
+                connection_id,
+            } => {
+                self.handle_send(&recipient, message, with_reply_surb, connection_id)
+                    .await
+            }
             ClientRequest::Reply {
                 message,
                 reply_surb,
-            } => self.handle_reply(reply_surb, message),
+            } => self.handle_reply(reply_surb, message).await,
             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),
         }
     }
 
-    fn handle_text_message(&mut self, msg: String) -> Option<WsMessage> {
+    async fn handle_text_message(&mut self, msg: String) -> Option<WsMessage> {
         debug!("Handling text message request");
         trace!("Content: {:?}", msg);
 
@@ -128,13 +208,13 @@ impl Handler {
 
         let response = match client_request {
             Err(err) => Some(ServerResponse::Error(err)),
-            Ok(req) => self.handle_request(req),
+            Ok(req) => self.handle_request(req).await,
         };
 
         response.map(|resp| WsMessage::text(resp.into_text()))
     }
 
-    fn handle_binary_message(&mut self, msg: Vec<u8>) -> Option<WsMessage> {
+    async fn handle_binary_message(&mut self, msg: &[u8]) -> Option<WsMessage> {
         debug!("Handling binary message request");
 
         self.received_response_type = ReceivedResponseType::Binary;
@@ -142,49 +222,23 @@ impl Handler {
 
         let response = match client_request {
             Err(err) => Some(ServerResponse::Error(err)),
-            Ok(req) => self.handle_request(req),
+            Ok(req) => self.handle_request(req).await,
         };
 
         response.map(|resp| WsMessage::Binary(resp.into_binary()))
     }
 
-    fn handle_ws_request(&mut self, raw_request: WsMessage) -> Option<WsMessage> {
+    async 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),
-            WsMessage::Binary(binary_message) => self.handle_binary_message(binary_message),
+            WsMessage::Text(text_message) => self.handle_text_message(text_message).await,
+            WsMessage::Binary(binary_message) => self.handle_binary_message(&binary_message).await,
             _ => 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>,
@@ -193,10 +247,8 @@ 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 => {
-                self.prepare_reconstructed_binary(reconstructed_messages)
-            }
-            ReceivedResponseType::Text => self.prepare_reconstructed_text(reconstructed_messages),
+            ReceivedResponseType::Binary => prepare_reconstructed_binary(reconstructed_messages),
+            ReceivedResponseType::Text => prepare_reconstructed_text(reconstructed_messages),
         };
 
         let mut send_stream = futures::stream::iter(response_messages);
@@ -244,7 +296,7 @@ impl Handler {
                         break;
                     }
 
-                    if let Some(response) = self.handle_ws_request(socket_msg) {
+                    if let Some(response) = self.handle_ws_request(socket_msg).await {
                         if let Err(err) = self.send_websocket_response(response).await {
                             warn!(
                                 "Failed to send message over websocket: {}. Assuming the connection is dead.",
@@ -291,3 +343,27 @@ 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,6 +20,12 @@ 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 {
@@ -28,32 +34,42 @@ 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 || data_len || data
-    fn serialize_send(recipient: Recipient, data: Vec<u8>, with_reply_surb: bool) -> Vec<u8> {
+    // SEND_REQUEST_TAG || with_surb || recipient || conn_id || data_len || data
+    fn serialize_send(
+        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 || data_len || data
+    // SEND_REQUEST_TAG || with_reply || recipient || conn_id || 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 + sizeof<u64> bytes
-        if b.len() < 2 + Recipient::LEN + size_of::<u64>() {
+        // we need to have at least 1 (tag) + 1 (reply flag) + Recipient::LEN + 2*sizeof<u64> bytes
+        if b.len() < 2 + Recipient::LEN + 2 * size_of::<u64>() {
             return Err(error::Error::new(
                 ErrorKind::TooShortRequest,
                 "not enough data provided to recover 'send'".to_string(),
@@ -86,9 +102,20 @@ impl ClientRequest {
             }
         };
 
-        let data_len_bytes = &b[2 + Recipient::LEN..2 + Recipient::LEN + size_of::<u64>()];
+        let mut connection_id_bytes = [0u8; 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 + size_of::<u64>()..];
+        let data = &b[2 + Recipient::LEN + 2 * size_of::<u64>()..];
         if data.len() as u64 != data_len {
             return Err(error::Error::new(
                 ErrorKind::MalformedRequest,
@@ -104,11 +131,12 @@ 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();
@@ -202,20 +230,79 @@ 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,
-            } => Self::serialize_send(recipient, message, with_reply_surb),
+                connection_id,
+            } => 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),
         }
     }
 
@@ -245,15 +332,17 @@ 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: Vec<u8>) -> Result<Self, error::Error> {
-        Self::deserialize(&raw_req)
+    pub fn try_from_binary(raw_req: &[u8]) -> Result<Self, error::Error> {
+        Self::deserialize(raw_req)
     }
 
     pub fn try_from_text(raw_req: String) -> Result<Self, error::Error> {
@@ -280,6 +369,7 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: false,
+            connection_id: Some(42),
         };
 
         let bytes = send_request_no_surb.serialize();
@@ -289,10 +379,12 @@ 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!(),
         }
@@ -301,6 +393,7 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: true,
+            connection_id: None,
         };
 
         let bytes = send_request_surb.serialize();
@@ -310,10 +403,12 @@ 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!(),
         }
@@ -352,4 +447,15 @@ 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,10 +23,14 @@ 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),
 }
 
@@ -193,6 +197,31 @@ 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();
@@ -272,6 +301,9 @@ 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),
         }
     }
@@ -302,6 +334,7 @@ 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,
@@ -378,6 +411,20 @@ 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,6 +20,7 @@ pub(super) enum ClientRequestText {
         message: String,
         recipient: String,
         with_reply_surb: bool,
+        connection_id: Option<u64>,
     },
     SelfAddress,
     #[serde(rename_all = "camelCase")]
@@ -46,6 +47,7 @@ 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| {
@@ -56,6 +58,7 @@ impl TryInto<ClientRequest> for ClientRequestText {
                     message: message_bytes,
                     recipient,
                     with_reply_surb,
+                    connection_id,
                 })
             }
             ClientRequestText::SelfAddress => Ok(ClientRequest::SelfAddress),
@@ -91,6 +94,10 @@ pub(super) enum ServerResponseText {
     SelfAddress {
         address: String,
     },
+    LaneQueueLength {
+        lane: u64,
+        queue_length: usize,
+    },
     Error {
         message: String,
     },
@@ -132,6 +139,9 @@ 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.0.2"
+version = "1.1.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"
@@ -20,11 +20,13 @@ pretty_env_logger = "0.4"
 rand = { version = "0.7.3", features = ["wasm-bindgen"] }
 serde = { version = "1.0", features = ["derive"] } # for config serialization/deserialization
 snafu = "0.6"
+thiserror = "1.0.34"
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal"] }
 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" }
@@ -44,6 +46,7 @@ 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,6 +23,13 @@ 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,13 +1,15 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::sync::atomic::Ordering;
+use std::error::Error;
 
 use crate::client::config::Config;
+use crate::error::Socks5ClientError;
 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,
@@ -23,6 +25,7 @@ use client_core::client::topology_control::{
     TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
 };
 use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
+use client_core::error::ClientCoreError;
 use crypto::asymmetric::identity;
 use futures::channel::mpsc;
 use futures::StreamExt;
@@ -34,7 +37,8 @@ use gateway_client::{
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::addressing::nodes::NodeIdentity;
-use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
+use tap::TapFallible;
+use task::{wait_for_signal_and_error, ShutdownListener, ShutdownNotifier};
 
 pub mod config;
 
@@ -108,6 +112,7 @@ impl NymClient {
         stream.start_with_shutdown(shutdown);
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn start_real_traffic_controller(
         &self,
         topology_accessor: TopologyAccessor,
@@ -115,6 +120,8 @@ 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(
@@ -144,6 +151,8 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
+            lane_queue_lengths,
+            client_connection_rx,
         )
         .start_with_shutdown(shutdown);
     }
@@ -190,11 +199,37 @@ impl NymClient {
             .expect("provided gateway id is invalid!");
 
         #[cfg(feature = "coconut")]
-        let bandwidth_controller = BandwidthController::new(
-            credential_storage::initialise_storage(self.config.get_base().get_database_path())
-                .await,
-            self.config.get_base().get_validator_api_endpoints(),
-        );
+        let bandwidth_controller = {
+            let details = network_defaults::NymNetworkDetails::new_from_env();
+            let mut client_config =
+                validator_client::Config::try_from_nym_network_details(&details)
+                    .expect("failed to construct validator client config");
+            let nymd_url = self
+                .config
+                .get_base()
+                .get_validator_endpoints()
+                .pop()
+                .expect("No nymd validator endpoint provided");
+            let api_url = self
+                .config
+                .get_base()
+                .get_validator_api_endpoints()
+                .pop()
+                .expect("No validator api endpoint provided");
+            // overwrite env configuration with config URLs
+            client_config = client_config.with_urls(nymd_url, api_url);
+            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())
@@ -232,7 +267,7 @@ impl NymClient {
         &mut self,
         topology_accessor: TopologyAccessor,
         shutdown: ShutdownListener,
-    ) {
+    ) -> Result<(), Socks5ClientError> {
         let topology_refresher_config = TopologyRefresherConfig::new(
             self.config.get_base().get_validator_api_endpoints(),
             self.config.get_base().get_topology_refresh_rate(),
@@ -247,14 +282,16 @@ impl NymClient {
 
         // TODO: a slightly more graceful termination here
         if !topology_refresher.is_topology_routable().await {
-            panic!(
-                "The current network topology seem to be insufficient to route any packets through\
+            log::error!(
+                "The current network topology seem to be insufficient to route any packets through \
                 - check if enough nodes and a gateway are online"
             );
+            return Err(ClientCoreError::InsufficientNetworkTopology.into());
         }
 
         info!("Starting topology refresher...");
         topology_refresher.start_with_shutdown(shutdown);
+        Ok(())
     }
 
     // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
@@ -275,7 +312,9 @@ impl NymClient {
         &self,
         buffer_requester: ReceivedBufferRequestSender,
         msg_input: InputMessageSender,
-        shutdown: ShutdownListener,
+        client_connection_tx: ConnectionCommandSender,
+        lane_queue_lengths: LaneQueueLengths,
+        mut shutdown: ShutdownListener,
     ) {
         info!("Starting socks5 listener...");
         let auth_methods = vec![AuthenticationMethods::NoAuth as u8];
@@ -287,30 +326,58 @@ impl NymClient {
             authenticator,
             self.config.get_provider_mix_address(),
             self.as_mix_recipient(),
-            shutdown,
+            lane_queue_lengths,
+            shutdown.clone(),
         );
-        tokio::spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
+        tokio::spawn(async move {
+            // 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) {
-        let mut shutdown = self.start().await;
-        wait_for_signal().await;
+    pub async fn run_forever(&mut self) -> Result<(), Box<dyn Error + Send>> {
+        let mut shutdown = self
+            .start()
+            .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
     }
 
     // Variant of `run_forever` that listends for remote control messages
-    pub async fn run_and_listen(&mut self, mut receiver: Socks5ControlMessageReceiver) {
-        let mut shutdown = self.start().await;
-        tokio::select! {
+    pub async fn run_and_listen(
+        &mut self,
+        mut receiver: Socks5ControlMessageReceiver,
+    ) -> Result<(), Box<dyn Error + Send>> {
+        // Start the main task
+        let mut shutdown = self
+            .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 {
@@ -321,23 +388,29 @@ 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
     }
 
-    pub async fn start(&mut self) -> ShutdownNotifier {
+    pub async fn start(&mut self) -> Result<ShutdownNotifier, Socks5ClientError> {
         info!("Starting nym client");
         // channels for inter-component communication
         // TODO: make the channels be internally created by the relevant components
@@ -352,7 +425,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) = mpsc::unbounded::<InputMessage>();
+        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
 
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
@@ -360,7 +433,10 @@ impl NymClient {
 
         let reply_key_storage =
             ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
-                .expect("Failed to load reply key storage!");
+                .tap_err(|err| {
+                    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();
@@ -368,7 +444,7 @@ impl NymClient {
         // 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(), shutdown.subscribe())
-            .await;
+            .await?;
         self.start_received_messages_buffer_controller(
             received_buffer_request_receiver,
             mixnet_messages_receiver,
@@ -387,12 +463,22 @@ 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(),
         );
 
@@ -411,12 +497,14 @@ impl NymClient {
         self.start_socks5_listener(
             received_buffer_request_sender,
             input_sender,
+            client_connection_tx,
+            shared_lane_queue_lengths,
             shutdown.subscribe(),
         );
 
         info!("Client startup finished!");
         info!("The address of this client is: {}", self.as_mix_recipient());
 
-        shutdown
+        Ok(shutdown)
     }
 }

clients/socks5/src/commands/init.rs

@@ -29,9 +29,13 @@ pub(crate) struct Init {
     #[clap(long)]
     force_register_gateway: bool,
 
-    /// Comma separated list of rest endpoints of the validators
+    /// Comma separated list of rest endpoints of the nymd validators
     #[clap(long)]
-    validators: Option<String>,
+    nymd_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)]
@@ -52,7 +56,8 @@ pub(crate) struct Init {
 impl From<Init> for OverrideConfig {
     fn from(init_config: Init) -> Self {
         OverrideConfig {
-            validators: init_config.validators,
+            nymd_validators: init_config.nymd_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,6 +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 clap::CommandFactory;
 use clap::{Parser, Subcommand};
@@ -50,7 +52,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(long)]
+    #[clap(short, long)]
     pub(crate) config_env_file: Option<std::path::PathBuf>,
 
     #[clap(subcommand)]
@@ -61,8 +63,10 @@ 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),
 
@@ -75,7 +79,8 @@ pub(crate) enum Commands {
 
 // Configuration that can be overridden.
 pub(crate) struct OverrideConfig {
-    validators: Option<String>,
+    nymd_validators: Option<String>,
+    api_validators: Option<String>,
     port: Option<u16>,
     fastmode: bool,
 
@@ -83,20 +88,30 @@ pub(crate) struct OverrideConfig {
     enabled_credentials_mode: bool,
 }
 
-pub(crate) async fn execute(args: &Cli) {
+pub(crate) async fn execute(args: &Cli) -> Result<(), Box<dyn Error + Send>> {
     let bin_name = "nym-socks5-client";
 
     match &args.command {
         Commands::Init(m) => init::execute(m).await,
-        Commands::Run(m) => run::execute(m).await,
+        Commands::Run(m) => run::execute(m).await?,
         Commands::Upgrade(m) => upgrade::execute(m),
         Commands::Completions(s) => s.generate(&mut Cli::into_app(), bin_name),
         Commands::GenerateFigSpec => fig_generate(&mut Cli::into_app(), bin_name),
     }
+    Ok(())
 }
 
 pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Config {
-    if let Some(raw_validators) = args.validators {
+    if let Some(raw_validators) = args.nymd_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

@@ -4,6 +4,7 @@
 use crate::{
     client::{config::Config, NymClient},
     commands::{override_config, OverrideConfig},
+    error::Socks5ClientError,
 };
 
 use clap::Args;
@@ -30,9 +31,13 @@ pub(crate) struct Run {
     #[clap(long)]
     gateway: Option<String>,
 
-    /// Comma separated list of rest endpoints of the validators
+    /// Comma separated list of rest endpoints of the nymd validators
     #[clap(long)]
-    validators: Option<String>,
+    nymd_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)]
@@ -48,7 +53,8 @@ pub(crate) struct Run {
 impl From<Run> for OverrideConfig {
     fn from(run_config: Run) -> Self {
         OverrideConfig {
-            validators: run_config.validators,
+            nymd_validators: run_config.nymd_validators,
+            api_validators: run_config.api_validators,
             port: run_config.port,
             fastmode: false,
 
@@ -80,14 +86,16 @@ fn version_check(cfg: &Config) -> bool {
     }
 }
 
-pub(crate) async fn execute(args: &Run) {
+pub(crate) async fn execute(args: &Run) -> Result<(), Box<dyn std::error::Error + Send>> {
     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;
+            return Err(Box::new(Socks5ClientError::FailedToLoadConfig(
+                id.to_string(),
+            )));
         }
     };
 
@@ -96,8 +104,8 @@ pub(crate) async fn execute(args: &Run) {
 
     if !version_check(&config) {
         error!("failed the local version check");
-        return;
+        return Err(Box::new(Socks5ClientError::FailedLocalVersionCheck));
     }
 
-    NymClient::new(config).run_forever().await;
+    NymClient::new(config).run_forever().await
 }

clients/socks5/src/error.rs

@@ -0,0 +1,32 @@
+use client_core::{client::reply_key_storage::ReplyKeyStorageError, 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}")]
+    IoError(#[from] std::io::Error),
+    #[error("Gateway client error: {0}")]
+    GatewayClientError(#[from] GatewayClientError),
+    #[error("Ed25519 error: {0}")]
+    Ed25519RecoveryError(#[from] Ed25519RecoveryError),
+    #[error("Validator client error: {0}")]
+    ValidatorClientError(#[from] ValidatorClientError),
+    #[error("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/lib.rs

@@ -2,4 +2,5 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod client;
+pub mod error;
 pub mod socks;

clients/socks5/src/main.rs

@@ -1,22 +1,25 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use std::error::Error;
+
 use clap::{crate_version, Parser};
 use logging::setup_logging;
 use network_defaults::setup_env;
 
 pub mod client;
 mod commands;
+pub mod error;
 pub mod socks;
 
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<(), Box<dyn Error + Send>> {
     setup_logging();
     println!("{}", banner());
 
     let args = commands::Cli::parse();
     setup_env(args.config_env_file.clone());
-    commands::execute(&args).await;
+    commands::execute(&args).await
 }
 
 fn banner() -> String {

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_core::client::inbound_messages::InputMessage;
-use client_core::client::inbound_messages::InputMessageSender;
+use client_connections::{LaneQueueLengths, TransmissionLane};
+use client_core::client::inbound_messages::{InputMessage, InputMessageSender};
 use futures::channel::mpsc;
 use futures::task::{Context, Poll};
 use log::*;
@@ -141,6 +141,7 @@ pub(crate) struct SocksClient {
     service_provider: Recipient,
     self_address: Recipient,
     started_proxy: bool,
+    lane_queue_lengths: LaneQueueLengths,
     shutdown_listener: ShutdownListener,
 }
 
@@ -158,6 +159,7 @@ impl Drop for SocksClient {
 
 impl SocksClient {
     /// Create a new SOCKClient
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         stream: TcpStream,
         authenticator: Authenticator,
@@ -165,9 +167,14 @@ impl SocksClient {
         service_provider: Recipient,
         controller_sender: ControllerSender,
         self_address: Recipient,
-        shutdown_listener: ShutdownListener,
+        lane_queue_lengths: LaneQueueLengths,
+        mut shutdown_listener: ShutdownListener,
     ) -> Self {
+        // If this task fails and exits, we don't want to send shutdown signal
+        shutdown_listener.mark_as_success();
+
         let connection_id = Self::generate_random();
+
         SocksClient {
             controller_sender,
             connection_id,
@@ -179,6 +186,7 @@ impl SocksClient {
             service_provider,
             self_address,
             started_proxy: false,
+            lane_queue_lengths,
             shutdown_listener,
         }
     }
@@ -198,6 +206,7 @@ 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(())
     }
 
@@ -230,8 +239,16 @@ 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(self.service_provider, msg.into_bytes(), false);
-        self.input_sender.unbounded_send(input_message).unwrap();
+        let input_message = InputMessage::new_fresh(
+            self.service_provider,
+            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) {
@@ -239,10 +256,15 @@ impl SocksClient {
             .await;
 
         let stream = self.stream.run_proxy();
-        let local_stream_remote = stream
-            .peer_addr()
-            .expect("failed to extract peer address")
-            .to_string();
+        let peer_addr = match stream.peer_addr() {
+            Ok(peer_addr) => peer_addr,
+            Err(err) => {
+                log::error!("Unable to extract the remote peer address: {err}");
+                return;
+            }
+        };
+        let local_stream_remote = peer_addr.to_string();
+
         let connection_id = self.connection_id;
         let input_sender = self.input_sender.clone();
 
@@ -254,12 +276,14 @@ 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);
-            InputMessage::new_fresh(recipient, provider_message.into_bytes(), false)
+            let lane = TransmissionLane::ConnectionId(conn_id);
+            InputMessage::new_fresh(recipient, provider_message.into_bytes(), false, lane)
         })
         .await
         .into_inner();

clients/socks5/src/socks/mixnet_responses.rs

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

clients/socks5/src/socks/server.rs

@@ -1,16 +1,17 @@
+use crate::error::Socks5ClientError;
+
 use super::authentication::Authenticator;
 use super::client::SocksClient;
-use super::{
-    mixnet_responses::MixnetResponseListener,
-    types::{ResponseCode, SocksProxyError},
-};
+use super::{mixnet_responses::MixnetResponseListener, types::ResponseCode};
+use client_connections::{ConnectionCommandSender, LaneQueueLengths};
 use client_core::client::{
     inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender,
 };
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
-use proxy_helpers::connection_controller::Controller;
+use proxy_helpers::connection_controller::{BroadcastActiveConnections, Controller};
 use std::net::SocketAddr;
+use tap::TapFallible;
 use task::ShutdownListener;
 use tokio::net::TcpListener;
 
@@ -20,6 +21,7 @@ pub struct SphinxSocksServer {
     listening_address: SocketAddr,
     service_provider: Recipient,
     self_address: Recipient,
+    lane_queue_lengths: LaneQueueLengths,
     shutdown: ShutdownListener,
 }
 
@@ -30,6 +32,7 @@ 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
@@ -41,6 +44,7 @@ impl SphinxSocksServer {
             listening_address: format!("{}:{}", ip, port).parse().unwrap(),
             service_provider,
             self_address,
+            lane_queue_lengths,
             shutdown,
         }
     }
@@ -51,13 +55,19 @@ impl SphinxSocksServer {
         &mut self,
         input_sender: InputMessageSender,
         buffer_requester: ReceivedBufferRequestSender,
-    ) -> Result<(), SocksProxyError> {
-        let listener = TcpListener::bind(self.listening_address).await.unwrap();
+        client_connection_tx: ConnectionCommandSender,
+    ) -> Result<(), Socks5ClientError> {
+        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(self.shutdown.clone());
+        let (mut active_streams_controller, controller_sender) = Controller::new(
+            client_connection_tx,
+            BroadcastActiveConnections::Off,
+            self.shutdown.clone(),
+        );
         tokio::spawn(async move {
             active_streams_controller.run().await;
         });
@@ -83,6 +93,7 @@ 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.0.0"
+version = "1.1.0"
 edition = "2021"
 keywords = ["nym", "sphinx", "wasm", "webassembly", "privacy", "client"]
 license = "Apache-2.0"
@@ -19,16 +19,18 @@ coconut = ["coconut-interface", "credentials", "gateway-client/coconut"]
 
 [dependencies]
 futures = "0.3"
-serde = { version = "1.0", features = ["derive"] }
-serde-wasm-bindgen = "0.4"
-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"] }
+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"
 
 # 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" }
@@ -55,6 +57,8 @@ wee_alloc = { version = "0.4", optional = true }
 wasm-bindgen-test = "0.3"
 
 [package.metadata.wasm-pack.profile.release]
-wasm-opt = false
-
+wasm-opt = true
 
+[profile.release]
+lto = true
+opt-level = 'z'

clients/webassembly/Makefile

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

clients/webassembly/js-example/webpack.config.js

@@ -2,6 +2,11 @@ const CopyWebpackPlugin = require('copy-webpack-plugin');
 const path = require('path');
 
 module.exports = {
+  performance: {
+    hints: false,
+    maxEntrypointSize: 512000,
+    maxAssetSize: 512000
+  },
   entry: {
     bootstrap: './bootstrap.js',
     worker: './worker.js',
@@ -22,6 +27,7 @@ module.exports = {
         },
       ],
     }),
+
   ],
   experiments: { syncWebAssembly: true },
 };

clients/webassembly/js-example/worker.js

@@ -73,6 +73,7 @@ async function main() {
   // const preferredGateway = 'CgQrYP8etksSBf4nALNqp93SHPpgFwEUyTsjBNNLj5WM';
 
   const gatewayEndpoint = await get_gateway(validator, preferredGateway);
+  gatewayEndpoint.gateway_listener = "wss://gateway1.nymtech.net:443"; // this is needed if we want it to work on the web. However this gateway is a v1 gateway, we will need to change for v2 once we get there
 
   // only really useful if you want to adjust some settings like traffic rate
   // (if not needed you can just pass a null)

clients/webassembly/src/binary_message_helper.rs

@@ -0,0 +1,217 @@
+// 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>) -> Vec<u8> {
+    create_binary_message_with_headers(kind, payload, "".to_string())
+}
+
+/// 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) -> Vec<u8> {
+    create_binary_message_with_headers(kind, payload.as_bytes().to_vec(), "".to_string())
+}
+
+/// 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: String) -> Vec<u8> {
+    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],
+            "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], "".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,6 +2,7 @@
 // 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},
@@ -25,7 +26,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, console_warn};
+use wasm_utils::console_log;
 
 pub mod config;
 
@@ -44,6 +45,7 @@ pub struct NymClient {
 
     // callbacks
     on_message: Option<js_sys::Function>,
+    on_binary_message: Option<js_sys::Function>,
     on_gateway_connect: Option<js_sys::Function>,
 }
 
@@ -55,6 +57,7 @@ impl NymClient {
             config,
             key_manager: Self::setup_key_manager(),
             on_message: None,
+            on_binary_message: None,
             on_gateway_connect: None,
             input_tx: None,
         }
@@ -72,8 +75,11 @@ 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)
     }
 
@@ -122,6 +128,8 @@ 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(),
@@ -146,6 +154,8 @@ impl NymClient {
             input_receiver,
             mix_sender,
             topology_accessor,
+            lane_queue_lengths,
+            client_connection_rx,
         )
         .start();
     }
@@ -266,6 +276,7 @@ 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();
@@ -280,8 +291,14 @@ impl NymClient {
             let this = JsValue::null();
 
             while let Some(reconstructed) = reconstructed_receiver.next().await {
-                if let Some(ref callback) = on_message {
-                    for msg in reconstructed {
+                for msg in reconstructed {
+                    if let Some(ref callback_binary) = on_binary_message {
+                        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)")
                         }
@@ -289,9 +306,6 @@ 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)
                 }
             }
         });
@@ -312,12 +326,16 @@ impl NymClient {
         let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
 
         // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = mpsc::unbounded::<InputMessage>();
+        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
 
         // 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())
@@ -337,11 +355,17 @@ 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 {
@@ -367,14 +391,16 @@ 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);
+        let input_msg = InputMessage::new_fresh(recipient, message, false, lane);
 
         self.input_tx
             .as_ref()
             .expect("start method was not called before!")
-            .unbounded_send(input_msg)
-            .unwrap();
+            .send(input_msg)
+            .await
+            .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::ApiClient::new(api_server.parse().unwrap());
+    let validator_client = validator_client::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,6 +3,8 @@
 
 use wasm_bindgen::prelude::*;
 
+#[cfg(target_arch = "wasm32")]
+pub mod binary_message_helper;
 #[cfg(target_arch = "wasm32")]
 mod client;
 #[cfg(target_arch = "wasm32")]

clippy.toml

@@ -0,0 +1,2 @@
+allow-unwrap-in-tests = true
+allow-expect-in-tests = true

common/client-connections/Cargo.toml

@@ -0,0 +1,10 @@
+[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

@@ -0,0 +1,111 @@
+// 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,6 +17,8 @@ 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::{
@@ -29,7 +31,7 @@ pub struct BandwidthController<St: Storage> {
     #[allow(dead_code)]
     storage: St,
     #[cfg(feature = "coconut")]
-    validator_endpoints: Vec<url::Url>,
+    coconut_api_clients: Vec<CoconutApiClient>,
 }
 
 impl<St> BandwidthController<St>
@@ -37,10 +39,10 @@ where
     St: Storage + Clone + 'static,
 {
     #[cfg(feature = "coconut")]
-    pub fn new(storage: St, validator_endpoints: Vec<url::Url>) -> Self {
+    pub fn new(storage: St, coconut_api_clients: Vec<CoconutApiClient>) -> Self {
         BandwidthController {
             storage,
-            validator_endpoints,
+            coconut_api_clients,
         }
     }
 
@@ -53,7 +55,7 @@ where
     pub async fn prepare_coconut_credential(
         &self,
     ) -> Result<coconut_interface::Credential, GatewayClientError> {
-        let verification_key = obtain_aggregate_verification_key(&self.validator_endpoints).await?;
+        let verification_key = obtain_aggregate_verification_key(&self.coconut_api_clients).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,25 +2,19 @@
 // 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};
-#[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 crate::{cleanup_socket_message, try_decrypt_binary_message};
 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};
+use gateway_requests::{BinaryRequest, ClientControlRequest, ServerResponse, PROTOCOL_VERSION};
 use log::*;
 use network_defaults::{REMAINING_BANDWIDTH_THRESHOLD, TOKENS_TO_BURN};
 use nymsphinx::forwarding::packet::MixPacket;
@@ -28,13 +22,20 @@ use rand::rngs::OsRng;
 use std::convert::TryFrom;
 use std::sync::Arc;
 use std::time::Duration;
-#[cfg(not(target_arch = "wasm32"))]
-use task::ShutdownListener;
 use tungstenite::protocol::Message;
 
+#[cfg(feature = "coconut")]
+use coconut_interface::Credential;
+
+#[cfg(not(target_arch = "wasm32"))]
+use credential_storage::PersistentStorage;
+#[cfg(not(target_arch = "wasm32"))]
+use task::ShutdownListener;
 #[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")]
@@ -306,6 +307,8 @@ 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
@@ -336,7 +339,15 @@ impl GatewayClient {
                     };
                     match ws_msg {
                         Message::Binary(bin_msg) => {
-                            if let Err(err) = self.packet_router.route_received(vec![bin_msg]) {
+                            // if we have established the shared key already, attempt to use it for decryption
+                            // 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);
                             }
                         }
@@ -436,6 +447,33 @@ 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);
@@ -458,11 +496,20 @@ impl GatewayClient {
             .map_err(GatewayClientError::RegistrationFailure),
             _ => unreachable!(),
         }?;
-        self.authenticated = match self.read_control_response().await? {
-            ServerResponse::Register { status } => Ok(status),
-            ServerResponse::Error { message } => Err(GatewayClientError::GatewayError(message)),
-            _ => Err(GatewayClientError::UnexpectedResponse),
-        }?;
+        let (authentication_status, gateway_protocol) = match self.read_control_response().await? {
+            ServerResponse::Register {
+                protocol_version,
+                status,
+            } => (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));
         }
@@ -501,9 +548,11 @@ 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,6 +85,9 @@ 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,6 +3,9 @@
 
 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,
 };
@@ -37,3 +40,21 @@ 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,14 +1,13 @@
 // 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"))]
@@ -50,21 +49,9 @@ impl PartiallyDelegated {
             match ws_msg {
                 Message::Binary(bin_msg) => {
                     // this function decrypts the request and checks the MAC
-                    let plaintext = match BinaryResponse::try_from_encrypted_tagged_bytes(
-                        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)
+                    if let Some(plaintext) = try_decrypt_binary_message(bin_msg, shared_key) {
+                        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,10 +10,11 @@ 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" }
@@ -37,6 +38,7 @@ 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 }
@@ -53,6 +55,7 @@ nymd-client = [
     "bip39",
     "config",
     "cosmrs",
+    "cw3",
     "prost",
     "flate2",
     "sha2",

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

@@ -1,14 +1,22 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // 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::families::{Family, FamilyHead};
 use mixnet_contract_common::mixnode::MixNodeDetails;
-use mixnet_contract_common::MixId;
 use mixnet_contract_common::{GatewayBond, IdentityKeyRef};
-use url::Url;
+use mixnet_contract_common::{IdentityKey, MixId};
+#[cfg(feature = "nymd-client")]
+use std::str::FromStr;
 use validator_api_requests::coconut::{
-    BlindSignRequestBody, BlindedSignatureResponse, CosmosAddressResponse, VerificationKeyResponse,
-    VerifyCredentialBody, VerifyCredentialResponse,
+    BlindSignRequestBody, BlindedSignatureResponse, VerifyCredentialBody, VerifyCredentialResponse,
 };
 use validator_api_requests::models::{
     GatewayCoreStatusResponse, MixnodeCoreStatusResponse, MixnodeStatusResponse,
@@ -16,10 +24,12 @@ use validator_api_requests::models::{
 };
 
 #[cfg(feature = "nymd-client")]
-use crate::nymd::traits::MixnetQueryClient;
+use crate::nymd::traits::{DkgQueryClient, MixnetQueryClient, MultisigQueryClient};
 #[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},
@@ -27,6 +37,7 @@ 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;
 
@@ -43,6 +54,9 @@ 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")]
@@ -72,6 +86,9 @@ 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,
         })
     }
 
@@ -119,6 +136,9 @@ 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,
@@ -145,6 +165,9 @@ 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,
         })
@@ -178,6 +201,9 @@ 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,
         })
@@ -194,6 +220,7 @@ impl Client<QueryNymdClient> {
 impl<C> Client<C> {
     // use case: somebody initialised client without a contract in order to upload and initialise one
     // and now they want to actually use it without making new client
+
     pub fn set_mixnet_contract_address(&mut self, mixnet_contract_address: cosmrs::AccountId) {
         self.nymd
             .set_mixnet_contract_address(mixnet_contract_address)
@@ -203,6 +230,56 @@ impl<C> Client<C> {
         self.nymd.mixnet_contract_address().clone()
     }
 
+    pub async fn get_all_node_families(&self) -> Result<Vec<Family>, ValidatorClientError>
+    where
+        C: CosmWasmClient + Sync + Send,
+    {
+        let mut families = Vec::new();
+        let mut start_after = None;
+
+        loop {
+            let paged_response = self
+                .nymd
+                .get_all_node_families_paged(start_after.take(), None)
+                .await?;
+            families.extend(paged_response.families);
+
+            if let Some(start_after_res) = paged_response.start_next_after {
+                start_after = Some(start_after_res)
+            } else {
+                break;
+            }
+        }
+
+        Ok(families)
+    }
+
+    pub async fn get_all_family_members(
+        &self,
+    ) -> Result<Vec<(IdentityKey, FamilyHead)>, ValidatorClientError>
+    where
+        C: CosmWasmClient + Sync + Send,
+    {
+        let mut members = Vec::new();
+        let mut start_after = None;
+
+        loop {
+            let paged_response = self
+                .nymd
+                .get_all_family_members_paged(start_after.take(), None)
+                .await?;
+            members.extend(paged_response.members);
+
+            if let Some(start_after_res) = paged_response.start_next_after {
+                start_after = Some(start_after_res)
+            } else {
+                break;
+            }
+        }
+
+        Ok(members)
+    }
+
     // basically handles paging for us
     pub async fn get_all_nymd_rewarded_set_mixnodes(
         &self,
@@ -519,6 +596,135 @@ 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
@@ -572,14 +778,53 @@ impl<C> Client<C> {
     ) -> Result<BlindedSignatureResponse, ValidatorClientError> {
         Ok(self.validator_api.blind_sign(request_body).await?)
     }
+}
 
-    pub async fn get_coconut_verification_key(
-        &self,
-    ) -> Result<VerificationKeyResponse, ValidatorClientError> {
-        Ok(self.validator_api.get_coconut_verification_key().await?)
+#[derive(Clone)]
+pub struct CoconutApiClient {
+    pub api_client: ApiClient,
+    pub verification_key: VerificationKey,
+    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)
@@ -685,16 +930,6 @@ 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,7 +9,8 @@ mod error;
 pub mod nymd;
 pub mod validator_api;
 
-pub use crate::client::ApiClient;
+#[cfg(feature = "nymd-client")]
+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,18 +6,21 @@ 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)]
 pub struct Log {
     #[serde(default)]
     // weird thing is that the first msg_index seems to always be undefined on the raw logs
-    msg_index: usize,
+    pub msg_index: usize,
     // unless I'm missing something obvious, the "log" type in cosmjs is always an empty string
     // and launchpad cosmos validator was setting it to what essentially is just the raw version of what
     // we received (and we don't care about launchpad, we, as the time of writing this, work on the stargate)
     // log: String,
-    events: Vec<cosmwasm_std::Event>,
+    pub events: Vec<cosmwasm_std::Event>,
 }
 
 /// Searches in logs for the first event of the given event type and in that event

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

@@ -65,6 +65,7 @@ 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,
 }
@@ -118,6 +119,10 @@ 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,
+            )?,
         })
     }
 }
@@ -275,6 +280,14 @@ 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

@@ -0,0 +1,126 @@
+// 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
+    }
+}