Made vesting_contract_address argument peroperly optional (#1769)

* Made vesting_contract_address argument peroperly optional

* Updated changelog

.github/workflows/audit.yml

@@ -1,36 +1,52 @@
 name: Daily security audit
 
-on: workflow_dispatch
+on:
+  schedule:
+    - cron: '5 9 * * *'
 jobs:
-  security_audit:
+  cargo-deny:
     runs-on: ubuntu-latest
     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
+    needs: cargo-deny
     runs-on: ubuntu-latest
     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_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/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-latest]
+
+    runs-on: ${{ matrix.platform }}
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Install Dependencies (Linux)
+        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
+
+      - name: Check the release tag starts with `nym-explorer-api-`
+        if: startsWith(github.ref, 'refs/tags/nym-explorer-api-') == false && github.event_name != 'workflow_dispatch'
+        uses: actions/github-script@v3
+        with:
+          script: |
+            core.setFailed('Release tag did not start with nym-explorer-api-...')
+
+      - name: Install Rust stable
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+
+      - name: Build all explorer-api
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --manifest-path explorer-api/Cargo.toml --workspace --release
+
+      - name: Upload Artifact
+        uses: actions/upload-artifact@v3
+        with:
+          name: my-artifact
+          path: |
+            target/release/explorer-api
+          retention-days: 30
+
+      - name: Upload to release based on tag name
+        uses: softprops/action-gh-release@v1
+        if: github.event_name == 'release'
+        with:
+          files: |
+            target/release/explorer-api

.github/workflows/nightly_build_release.yml

@@ -1,4 +1,4 @@
-name: Nightly builds on dispatch
+name: Nightly builds on latest release
 
 on: workflow_dispatch
 jobs:
@@ -7,26 +7,40 @@ jobs:
     outputs:
       matrix: ${{ steps.set-matrix.outputs.matrix }}
     steps:
-      # creates the matrix strategy from nightly_build_matrix_includes.json
-      - uses: actions/checkout@v2
+      # creates the matrix strategy from nightly_build_release_matrix.json
+      - 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_release_matrix.json'
           filter: '[?runOnEvent==`${{ github.event_name }}` || runOnEvent==`always`]'
-  build:
+  get_release:
+    runs-on: ubuntu-latest
     needs: matrix_prep
+    outputs:
+      output1: ${{ steps.step2.outputs.lastest_release }}
+    steps:
+      - name: Check out repository code
+        uses: actions/checkout@v3
+
+      - 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
+        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
         if: matrix.os == 'ubuntu-latest'
 
-      - name: Check out 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 +56,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:
@@ -99,6 +119,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:
@@ -151,7 +177,7 @@ jobs:
       - 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 +186,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##*/}"
           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/nym-release-publish.yml

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

.github/workflows/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,
+};

CHANGELOG.md

@@ -6,34 +6,49 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ### 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])
+- binaries: add `-c` shortform for `--config-env-file`
 
 ### 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])
+- clients: add concept of transmission lanes to better handle multiple data streams ([#1720])
 
+[#1720]: https://github.com/nymtech/nym/pull/1720
+
+
+## [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 +56,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,17 @@ dependencies = [
 ]
 
 [[package]]
-name = "client-core"
-version = "1.0.1"
+name = "client-connections"
+version = "0.1.0"
 dependencies = [
+ "futures",
+]
+
+[[package]]
+name = "client-core"
+version = "1.1.0"
+dependencies = [
+ "client-connections",
  "config",
  "crypto",
  "dirs",
@@ -850,9 +858,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",
 ]
@@ -1582,7 +1590,7 @@ dependencies = [
 
 [[package]]
 name = "explorer-api"
-version = "1.0.1"
+version = "1.1.0"
 dependencies = [
  "chrono",
  "clap 3.2.8",
@@ -3070,7 +3078,7 @@ dependencies = [
 
 [[package]]
 name = "nym-cli"
-version = "1.0.0"
+version = "1.1.0"
 dependencies = [
  "anyhow",
  "base64",
@@ -3122,9 +3130,10 @@ dependencies = [
 
 [[package]]
 name = "nym-client"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "clap 3.2.8",
+ "client-connections",
  "client-core",
  "coconut-interface",
  "completions",
@@ -3147,6 +3156,7 @@ dependencies = [
  "serde_json",
  "sled",
  "task",
+ "thiserror",
  "tokio",
  "tokio-tungstenite 0.14.0",
  "topology",
@@ -3159,7 +3169,7 @@ dependencies = [
 
 [[package]]
 name = "nym-gateway"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3206,7 +3216,7 @@ dependencies = [
 
 [[package]]
 name = "nym-mixnode"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "anyhow",
  "bs58",
@@ -3248,10 +3258,11 @@ dependencies = [
 
 [[package]]
 name = "nym-network-requester"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "async-trait",
  "clap 3.2.8",
+ "client-connections",
  "completions",
  "dirs",
  "futures",
@@ -3295,9 +3306,10 @@ dependencies = [
 
 [[package]]
 name = "nym-socks5-client"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "clap 3.2.8",
+ "client-connections",
  "client-core",
  "coconut-interface",
  "completions",
@@ -3323,6 +3335,7 @@ dependencies = [
  "snafu 0.6.10",
  "socks5-requests",
  "task",
+ "thiserror",
  "tokio",
  "topology",
  "url",
@@ -3359,7 +3372,7 @@ dependencies = [
 
 [[package]]
 name = "nym-validator-api"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -4121,6 +4134,7 @@ name = "proxy-helpers"
 version = "0.1.0"
 dependencies = [
  "bytes",
+ "client-connections",
  "futures",
  "log",
  "ordered-buffer",
@@ -6514,7 +6528,6 @@ version = "0.1.0"
 dependencies = [
  "contracts-common",
  "cosmwasm-std",
- "log",
  "mixnet-contract-common",
  "schemars",
  "serde",

Cargo.toml

@@ -26,6 +26,7 @@ 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",

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.0"
 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");

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

@@ -1,3 +1,4 @@
+use client_connections::TransmissionLane;
 use futures::channel::mpsc;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::anonymous_replies::ReplySurb;
@@ -11,6 +12,7 @@ pub enum InputMessage {
         recipient: Recipient,
         data: Vec<u8>,
         with_reply_surb: bool,
+        lane: TransmissionLane,
     },
     Reply {
         reply_surb: ReplySurb,
@@ -19,11 +21,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/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)
         {

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

@@ -8,6 +8,7 @@ use crate::client::{
     real_messages_control::real_traffic_stream::{BatchRealMessageSender, RealMessage},
     topology_control::TopologyAccessor,
 };
+use client_connections::TransmissionLane;
 use futures::StreamExt;
 use log::*;
 use nymsphinx::anonymous_replies::ReplySurb;
@@ -104,6 +105,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,26 +166,30 @@ 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)
+                .unbounded_send((real_messages, lane))
                 .unwrap();
         }
     }

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,10 +117,10 @@ 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,
-            )])
+            .unbounded_send((
+                vec![RealMessage::new(prepared_fragment.mix_packet, frag_id)],
+                TransmissionLane::Retransmission,
+            ))
             .unwrap();
     }
 

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

@@ -14,6 +14,7 @@ use crate::client::{
     topology_control::TopologyAccessor,
 };
 use crate::spawn_future;
+use client_connections::ClosedConnectionReceiver;
 use futures::channel::mpsc;
 use gateway_client::AcknowledgementReceiver;
 use log::*;
@@ -110,6 +111,7 @@ impl RealMessagesController<OsRng> {
         mix_sender: BatchMixMessageSender,
         topology_access: TopologyAccessor,
         #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
+        closed_connection_rx: ClosedConnectionReceiver,
     ) -> Self {
         let rng = OsRng;
 
@@ -159,6 +161,7 @@ impl RealMessagesController<OsRng> {
             rng,
             config.self_recipient,
             topology_access,
+            closed_connection_rx,
         );
 
         RealMessagesController {

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

@@ -4,6 +4,7 @@
 use crate::client::mix_traffic::BatchMixMessageSender;
 use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender;
 use crate::client::topology_control::TopologyAccessor;
+use client_connections::{ClosedConnectionReceiver, ConnectionId, TransmissionLane};
 use futures::channel::mpsc;
 use futures::task::{Context, Poll};
 use futures::{Future, Stream, StreamExt};
@@ -16,7 +17,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 +27,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 +85,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 +108,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,8 +127,13 @@ 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
+    closed_connection_rx: ClosedConnectionReceiver,
 }
 
 pub(crate) struct RealMessage {
@@ -242,8 +152,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 =
+    mpsc::UnboundedSender<(Vec<RealMessage>, TransmissionLane)>;
+type BatchRealMessageReceiver = mpsc::UnboundedReceiver<(Vec<RealMessage>, TransmissionLane)>;
 
 pub(crate) enum StreamMessage {
     Cover,
@@ -266,22 +177,21 @@ where
         rng: R,
         our_full_destination: Recipient,
         topology_access: TopologyAccessor,
+        closed_connection_rx: ClosedConnectionReceiver,
     ) -> 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(),
+            closed_connection_rx,
         }
     }
 
@@ -296,7 +206,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,20 +225,24 @@ 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))
             }
         };
 
@@ -336,6 +250,16 @@ where
             log::error!("Failed to send - channel closed: {}", 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 closed_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,35 +271,41 @@ 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();
         }
     }
 
@@ -385,6 +315,13 @@ where
         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.closed_connection_rx).poll_next(cx) {
+            self.on_close_connection(id);
+        }
+
         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 +346,35 @@ 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
+            // 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_next(cx) {
                 // in the case our real message channel stream was closed, we should also indicate we are closed
                 // (and whoever is using the stream should panic)
                 Poll::Ready(None) => Poll::Ready(None),
 
-                // 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
+                        .transmission_buffer
+                        .pop_next_message_at_random()
+                        .expect("we just added 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.transmission_buffer.pop_next_message_at_random() {
+                        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,14 +396,9 @@ 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()
-            }
-            return Poll::Ready(Some(StreamMessage::Real(Box::new(real_available))));
+        // Start by checking if we have any incoming messages about closed connections
+        if let Poll::Ready(Some(id)) = Pin::new(&mut self.closed_connection_rx).poll_next(cx) {
+            self.on_close_connection(id);
         }
 
         match Pin::new(&mut self.real_receiver).poll_next(cx) {
@@ -467,17 +406,26 @@ where
             // (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
+                    .transmission_buffer
+                    .pop_next_message_at_random()
+                    .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.transmission_buffer.pop_next_message_at_random() {
+                    Poll::Ready(Some(StreamMessage::Real(Box::new(real_next))))
+                } else {
+                    Poll::Pending
+                }
+            }
         }
     }
 
@@ -492,24 +440,47 @@ 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();
+        if self.config.disable_poisson_packet_distribution {
+            log::info!(
+                "Status: {lanes} lanes, backlog: {:.2} kiB ({packets}), no delay",
+                backlog
+            );
+        } else {
+            log::info!(
+                "Status: {lanes} lanes, backlog: {:.2} kiB ({packets}), avg delay: {}ms ({mult})",
+                backlog,
+                delay
+            );
+        }
+    }
+
     #[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(1));
+
         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();
+                }
+                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;
                 }
             }
         }

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::debug!(
+                "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,207 @@
+// 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()
+    }
+
+    #[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<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()?
+        };
+
+        log::trace!("picking to send from lane: {:?}", lane);
+        self.pop_front_from_lane(&lane)
+    }
+
+    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)
+    }
+
+    #[cfg(not(target_arch = "wasm32"))]
+    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()
         );

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
+                })
+        })
     }
 }
 

clients/client-core/src/error.rs

@@ -24,4 +24,6 @@ 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,
 }

clients/credential/src/main.rs

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

clients/native/Cargo.toml

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

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: 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: 0,
     };
 
     println!("sending content of 'dummy_file' over the mix network...");

clients/native/src/client/mod.rs

@@ -1,6 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use client_connections::{ClosedConnectionReceiver, ClosedConnectionSender, TransmissionLane};
 use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
 use client_core::client::inbound_messages::{
     InputMessage, InputMessageReceiver, InputMessageSender,
@@ -18,6 +19,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;
@@ -33,6 +35,7 @@ use nymsphinx::receiver::ReconstructedMessage;
 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 +111,7 @@ impl NymClient {
         stream.start_with_shutdown(shutdown);
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn start_real_traffic_controller(
         &self,
         topology_accessor: TopologyAccessor,
@@ -115,6 +119,7 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
+        closed_connection_rx: ClosedConnectionReceiver,
         shutdown: ShutdownListener,
     ) {
         let mut controller_config = real_messages_control::Config::new(
@@ -144,6 +149,7 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
+            closed_connection_rx,
         )
         .start_with_shutdown(shutdown);
     }
@@ -232,7 +238,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 +253,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 +283,16 @@ impl NymClient {
         &self,
         buffer_requester: ReceivedBufferRequestSender,
         msg_input: InputMessageSender,
+        closed_connection_tx: ClosedConnectionSender,
     ) {
         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,
+            closed_connection_tx,
+            buffer_requester,
+            &self.as_mix_recipient(),
+        );
 
         websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler);
     }
@@ -288,7 +301,8 @@ impl NymClient {
     /// 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);
+        let lane = TransmissionLane::General;
+        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb, lane);
 
         self.input_tx
             .as_ref()
@@ -328,8 +342,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 +360,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
@@ -379,7 +394,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 +413,17 @@ 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 (closed_connection_tx, closed_connection_rx) = mpsc::unbounded();
+
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             reply_key_storage,
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
+            closed_connection_rx,
             shutdown.subscribe(),
         );
 
@@ -420,9 +440,11 @@ 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,
+                closed_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 +465,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/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)]
@@ -83,16 +84,17 @@ 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 {

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;
@@ -73,14 +74,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 +90,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,23 @@
+use client_core::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("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,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use client_connections::{ClosedConnectionSender, TransmissionLane};
 use client_core::client::{
     inbound_messages::{InputMessage, InputMessageSender},
     received_buffer::{
@@ -34,6 +35,7 @@ impl Default for ReceivedResponseType {
 
 pub(crate) struct Handler {
     msg_input: InputMessageSender,
+    closed_connection_tx: ClosedConnectionSender,
     buffer_requester: ReceivedBufferRequestSender,
     self_full_address: Recipient,
     socket: Option<WebSocketStream<TcpStream>>,
@@ -45,6 +47,7 @@ impl Clone for Handler {
     fn clone(&self) -> Self {
         Handler {
             msg_input: self.msg_input.clone(),
+            closed_connection_tx: self.closed_connection_tx.clone(),
             buffer_requester: self.buffer_requester.clone(),
             self_full_address: self.self_full_address,
             socket: None,
@@ -64,13 +67,15 @@ impl Drop for Handler {
 impl Handler {
     pub(crate) fn new(
         msg_input: InputMessageSender,
+        closed_connection_tx: ClosedConnectionSender,
         buffer_requester: ReceivedBufferRequestSender,
-        self_full_address: Recipient,
+        self_full_address: &Recipient,
     ) -> Self {
         Handler {
             msg_input,
+            closed_connection_tx,
             buffer_requester,
-            self_full_address,
+            self_full_address: *self_full_address,
             socket: None,
             received_response_type: Default::default(),
         }
@@ -78,12 +83,14 @@ impl Handler {
 
     fn handle_send(
         &mut self,
-        recipient: Recipient,
+        recipient: &Recipient,
         message: Vec<u8>,
         with_reply_surb: bool,
+        connection_id: u64,
     ) -> Option<ServerResponse> {
         // the ack control is now responsible for chunking, etc.
-        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb);
+        let lane = TransmissionLane::ConnectionId(connection_id);
+        let input_msg = InputMessage::new_fresh(*recipient, message, with_reply_surb, lane);
         self.msg_input.unbounded_send(input_msg).unwrap();
 
         None
@@ -104,18 +111,27 @@ impl Handler {
         ServerResponse::SelfAddress(self.self_full_address)
     }
 
+    fn handle_closed_connection(&self, connection_id: u64) -> Option<ServerResponse> {
+        self.closed_connection_tx
+            .unbounded_send(connection_id)
+            .unwrap();
+        None
+    }
+
     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),
             ClientRequest::Reply {
                 message,
                 reply_surb,
             } => self.handle_reply(reply_surb, message),
             ClientRequest::SelfAddress => Some(self.handle_self_address()),
+            ClientRequest::ClosedConnection(id) => self.handle_closed_connection(id),
         }
     }
 
@@ -134,7 +150,7 @@ impl Handler {
         response.map(|resp| WsMessage::text(resp.into_text()))
     }
 
-    fn handle_binary_message(&mut self, msg: Vec<u8>) -> Option<WsMessage> {
+    fn handle_binary_message(&mut self, msg: &[u8]) -> Option<WsMessage> {
         debug!("Handling binary message request");
 
         self.received_response_type = ReceivedResponseType::Binary;
@@ -154,37 +170,11 @@ impl Handler {
         // 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::Binary(binary_message) => self.handle_binary_message(&binary_message),
             _ => None,
         }
     }
 
-    // I'm still not entirely sure why `send_all` requires `TryStream` rather than `Stream`, but
-    // let's just play along for now
-    fn prepare_reconstructed_binary(
-        &self,
-        reconstructed_messages: Vec<ReconstructedMessage>,
-    ) -> Vec<Result<WsMessage, WsError>> {
-        reconstructed_messages
-            .into_iter()
-            .map(ServerResponse::Received)
-            .map(|resp| Ok(WsMessage::Binary(resp.into_binary())))
-            .collect()
-    }
-
-    // I'm still not entirely sure why `send_all` requires `TryStream` rather than `Stream`, but
-    // let's just play along for now
-    fn prepare_reconstructed_text(
-        &self,
-        reconstructed_messages: Vec<ReconstructedMessage>,
-    ) -> Vec<Result<WsMessage, WsError>> {
-        reconstructed_messages
-            .into_iter()
-            .map(ServerResponse::Received)
-            .map(|resp| Ok(WsMessage::Text(resp.into_text())))
-            .collect()
-    }
-
     async fn push_websocket_received_plaintexts(
         &mut self,
         reconstructed_messages: Vec<ReconstructedMessage>,
@@ -193,10 +183,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);
@@ -291,3 +279,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,9 @@ 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;
+
 #[allow(non_snake_case)]
 #[derive(Debug)]
 pub enum ClientRequest {
@@ -28,32 +31,41 @@ 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: u64,
     },
     Reply {
         message: Vec<u8>,
         reply_surb: ReplySurb,
     },
     SelfAddress,
+    ClosedConnection(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: u64,
+    ) -> Vec<u8> {
         let data_len_bytes = (data.len() as u64).to_be_bytes();
+        let conn_id_bytes = connection_id.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 +98,15 @@ 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 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 +122,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 +221,43 @@ 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]) -> Self {
+        // 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);
+
+        ClientRequest::ClosedConnection(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),
         }
     }
 
@@ -245,15 +287,16 @@ 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 => Ok(Self::deserialize_closed_connection(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 +323,7 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: false,
+            connection_id: 42,
         };
 
         let bytes = send_request_no_surb.serialize();
@@ -289,10 +333,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, 42)
             }
             _ => unreachable!(),
         }
@@ -301,6 +347,7 @@ mod tests {
             recipient,
             message: b"foomp".to_vec(),
             with_reply_surb: true,
+            connection_id: 213,
         };
 
         let bytes = send_request_surb.serialize();
@@ -310,10 +357,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, 213)
             }
             _ => unreachable!(),
         }
@@ -352,4 +401,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/text.rs

@@ -20,6 +20,7 @@ pub(super) enum ClientRequestText {
         message: String,
         recipient: String,
         with_reply_surb: bool,
+        connection_id: 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),

clients/socks5/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-socks5-client"
-version = "1.0.2"
+version = "1.1.0"
 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" }

clients/socks5/src/client/mod.rs

@@ -4,10 +4,12 @@
 use std::sync::atomic::Ordering;
 
 use crate::client::config::Config;
+use crate::error::Socks5ClientError;
 use crate::socks::{
     authentication::{AuthenticationMethods, Authenticator, User},
     server::SphinxSocksServer,
 };
+use client_connections::{ClosedConnectionReceiver, ClosedConnectionSender};
 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;
@@ -108,6 +111,7 @@ impl NymClient {
         stream.start_with_shutdown(shutdown);
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn start_real_traffic_controller(
         &self,
         topology_accessor: TopologyAccessor,
@@ -115,6 +119,7 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
+        closed_connection_rx: ClosedConnectionReceiver,
         shutdown: ShutdownListener,
     ) {
         let mut controller_config = client_core::client::real_messages_control::Config::new(
@@ -144,6 +149,7 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
+            closed_connection_rx,
         )
         .start_with_shutdown(shutdown);
     }
@@ -232,7 +238,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 +253,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,6 +283,7 @@ impl NymClient {
         &self,
         buffer_requester: ReceivedBufferRequestSender,
         msg_input: InputMessageSender,
+        closed_connection_tx: ClosedConnectionSender,
         shutdown: ShutdownListener,
     ) {
         info!("Starting socks5 listener...");
@@ -289,12 +298,16 @@ impl NymClient {
             self.as_mix_recipient(),
             shutdown,
         );
-        tokio::spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
+        tokio::spawn(async move {
+            sphinx_socks
+                .serve(msg_input, buffer_requester, closed_connection_tx)
+                .await
+        });
     }
 
     /// 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;
+    pub async fn run_forever(&mut self) -> Result<(), Socks5ClientError> {
+        let mut shutdown = self.start().await?;
         wait_for_signal().await;
 
         log::info!("Sending shutdown");
@@ -305,11 +318,15 @@ impl NymClient {
         shutdown.wait_for_shutdown().await;
 
         log::info!("Stopping nym-socks5-client");
+        Ok(())
     }
 
     // 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;
+    pub async fn run_and_listen(
+        &mut self,
+        mut receiver: Socks5ControlMessageReceiver,
+    ) -> Result<(), Socks5ClientError> {
+        let mut shutdown = self.start().await?;
         tokio::select! {
             message = receiver.next() => {
                 log::debug!("Received message: {:?}", message);
@@ -335,9 +352,10 @@ impl NymClient {
         shutdown.wait_for_shutdown().await;
 
         log::info!("Stopping nym-socks5-client");
+        Ok(())
     }
 
-    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
@@ -368,7 +386,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 +405,17 @@ 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 (closed_connection_tx, closed_connection_rx) = mpsc::unbounded();
+
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             reply_key_storage,
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
+            closed_connection_rx,
             shutdown.subscribe(),
         );
 
@@ -411,12 +434,13 @@ impl NymClient {
         self.start_socks5_listener(
             received_buffer_request_sender,
             input_sender,
+            closed_connection_tx,
             shutdown.subscribe(),
         );
 
         info!("Client startup finished!");
         info!("The address of this client is: {}", self.as_mix_recipient());
 
-        shutdown
+        Ok(shutdown)
     }
 }

clients/socks5/src/commands/mod.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::client::config::Config;
+use crate::error::Socks5ClientError;
 use clap::CommandFactory;
 use clap::{Parser, Subcommand};
 use completions::{fig_generate, ArgShell};
@@ -50,7 +51,7 @@ fn long_version_static() -> &'static str {
 #[clap(author = "Nymtech", version, long_version = long_version_static(), about)]
 pub(crate) struct Cli {
     /// Path pointing to an env file that configures the client.
-    #[clap(long)]
+    #[clap(short, long)]
     pub(crate) config_env_file: Option<std::path::PathBuf>,
 
     #[clap(subcommand)]
@@ -61,8 +62,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),
 
@@ -83,16 +86,17 @@ pub(crate) struct OverrideConfig {
     enabled_credentials_mode: bool,
 }
 
-pub(crate) async fn execute(args: &Cli) {
+pub(crate) async fn execute(args: &Cli) -> Result<(), Socks5ClientError> {
     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 {

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;
@@ -80,14 +81,14 @@ fn version_check(cfg: &Config) -> bool {
     }
 }
 
-pub(crate) async fn execute(args: &Run) {
+pub(crate) async fn execute(args: &Run) -> Result<(), Socks5ClientError> {
     let id = &args.id;
 
     let mut config = match Config::load_from_file(Some(id)) {
         Ok(cfg) => cfg,
         Err(err) => {
             error!("Failed to load config for {}. Are you sure you have run `init` before? (Error was: {})", id, err);
-            return;
+            return Err(Socks5ClientError::FailedToLoadConfig(id.to_string()));
         }
     };
 
@@ -96,8 +97,8 @@ pub(crate) async fn execute(args: &Run) {
 
     if !version_check(&config) {
         error!("failed the local version check");
-        return;
+        return Err(Socks5ClientError::FailedLocalVersionCheck);
     }
 
-    NymClient::new(config).run_forever().await;
+    NymClient::new(config).run_forever().await
 }

clients/socks5/src/error.rs

@@ -0,0 +1,23 @@
+use client_core::error::ClientCoreError;
+use crypto::asymmetric::identity::Ed25519RecoveryError;
+use gateway_client::error::GatewayClientError;
+use validator_client::ValidatorClientError;
+
+#[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("Failed to load config for: {0}")]
+    FailedToLoadConfig(String),
+    #[error("Failed local version check, client and config mismatch")]
+    FailedLocalVersionCheck,
+}

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

@@ -2,21 +2,23 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::{crate_version, Parser};
+use error::Socks5ClientError;
 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<(), Socks5ClientError> {
     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::TransmissionLane;
+use client_core::client::inbound_messages::{InputMessage, InputMessageSender};
 use futures::channel::mpsc;
 use futures::task::{Context, Poll};
 use log::*;
@@ -226,17 +226,21 @@ impl SocksClient {
         }
     }
 
-    async fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) {
+    fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) {
         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);
+        let input_message = InputMessage::new_fresh(
+            self.service_provider,
+            msg.into_bytes(),
+            false,
+            TransmissionLane::ConnectionId(self.connection_id),
+        );
         self.input_sender.unbounded_send(input_message).unwrap();
     }
 
     async fn run_proxy(&mut self, conn_receiver: ConnectionReceiver, remote_proxy_target: String) {
-        self.send_connect_to_mixnet(remote_proxy_target.clone())
-            .await;
+        self.send_connect_to_mixnet(remote_proxy_target.clone());
 
         let stream = self.stream.run_proxy();
         let local_stream_remote = stream
@@ -259,7 +263,8 @@ impl SocksClient {
         .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/server.rs

@@ -4,6 +4,7 @@ use super::{
     mixnet_responses::MixnetResponseListener,
     types::{ResponseCode, SocksProxyError},
 };
+use client_connections::ClosedConnectionSender;
 use client_core::client::{
     inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender,
 };
@@ -51,13 +52,14 @@ impl SphinxSocksServer {
         &mut self,
         input_sender: InputMessageSender,
         buffer_requester: ReceivedBufferRequestSender,
+        closed_connection_tx: ClosedConnectionSender,
     ) -> Result<(), SocksProxyError> {
         let listener = TcpListener::bind(self.listening_address).await.unwrap();
         info!("Serving Connections...");
 
         // controller for managing all active connections
         let (mut active_streams_controller, controller_sender) =
-            Controller::new(self.shutdown.clone());
+            Controller::new(closed_connection_tx, self.shutdown.clone());
         tokio::spawn(async move {
             active_streams_controller.run().await;
         });

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.0.1"
 edition = "2021"
 keywords = ["nym", "sphinx", "wasm", "webassembly", "privacy", "client"]
 license = "Apache-2.0"
@@ -29,6 +29,7 @@ url = "2.2"
 
 # internal
 client-core = { path = "../client-core", default-features = false, features = ["wasm"] }
+client-connections = { path = "../../common/client-connections" }
 coconut-interface = { path = "../../common/coconut-interface", optional = true }
 credentials = { path = "../../common/credentials", optional = true }
 crypto = { path = "../../common/crypto" }
@@ -55,6 +56,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::{ClosedConnectionReceiver, 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,7 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
+        closed_connection_rx: ClosedConnectionReceiver,
     ) {
         let mut controller_config = real_messages_control::Config::new(
             self.key_manager.ack_key(),
@@ -146,6 +153,7 @@ impl NymClient {
             input_receiver,
             mix_sender,
             topology_accessor,
+            closed_connection_rx,
         )
         .start();
     }
@@ -266,6 +274,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 +289,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 +304,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)
                 }
             }
         });
@@ -318,6 +330,10 @@ impl NymClient {
         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 (_closed_connection_tx, closed_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())
@@ -342,6 +358,7 @@ impl NymClient {
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
+            closed_connection_rx,
         );
 
         if !self.config.debug.disable_loop_cover_traffic_stream {
@@ -367,8 +384,9 @@ 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()

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,9 @@
+[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"

common/client-connections/src/lib.rs

@@ -0,0 +1,21 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use futures::channel::mpsc;
+
+pub type ConnectionId = u64;
+
+#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)]
+pub enum TransmissionLane {
+    General,
+    Reply,
+    Retransmission,
+    Control,
+    ConnectionId(ConnectionId),
+}
+
+/// Announce connections that are closed, for whoever is interested.
+/// One usecase is that the network-requester and socks5-client wants to know about this, so that
+/// they can forward this to the `OutQueueControl` (via `ClientRequest` for the network-requester)
+pub type ClosedConnectionSender = mpsc::UnboundedSender<ConnectionId>;
+pub type ClosedConnectionReceiver = mpsc::UnboundedReceiver<ConnectionId>;

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

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

common/cosmwasm-smart-contracts/contracts-common/src/events.rs

@@ -11,6 +11,8 @@ use cosmwasm_std::Event;
 /// * `event`: event to search through.
 /// * `key`: key associated with the particular attribute
 pub fn must_find_attribute(event: &Event, key: &str) -> String {
+    // due to how the function is supposed to work, the unwrap is fine in this instance
+    #[allow(clippy::unwrap_used)]
     may_find_attribute(event, key).unwrap()
 }
 

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

@@ -1,6 +1,9 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+#![warn(clippy::expect_used)]
+#![warn(clippy::unwrap_used)]
+
 pub mod events;
 pub mod types;
 

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

@@ -72,7 +72,7 @@ impl Percent {
 
 impl Display for Percent {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        let adjusted = Decimal::from_atomics(100u32, 0).unwrap() * self.0;
+        let adjusted = Decimal::from_ratio(100u32, 1u32) * self.0;
         write!(f, "{}%", adjusted)
     }
 }

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

@@ -1,7 +1,9 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use cosmwasm_std::Decimal;
+use cosmwasm_std::{Decimal, Uint128};
+
+pub const TOKEN_SUPPLY: Uint128 = Uint128::new(1_000_000_000_000_000);
 
 // I'm still not 100% sure how to feel about existence of this file
 // This is equivalent of representing our display coin with 6 decimal places.

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

@@ -4,8 +4,10 @@
 // due to code generated by JsonSchema
 #![allow(clippy::field_reassign_with_default)]
 
+use crate::constants::TOKEN_SUPPLY;
+use crate::helpers::IntoBaseDecimal;
 use crate::{Addr, MixId};
-use cosmwasm_std::{Coin, Decimal};
+use cosmwasm_std::{Coin, Decimal, StdResult};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
@@ -60,6 +62,11 @@ impl Delegation {
         height: u64,
         proxy: Option<Addr>,
     ) -> Self {
+        assert!(
+            amount.amount <= TOKEN_SUPPLY,
+            "delegation cannot be larger than the token supply"
+        );
+
         Delegation {
             owner,
             mix_id,
@@ -87,10 +94,8 @@ impl Delegation {
         (mix_id, owner_proxy_subkey)
     }
 
-    pub fn dec_amount(&self) -> Decimal {
-        // the unwrap here is fine as we're guaranteed our base coin amount is going to fit in a Decimal
-        // with 0 decimal places
-        Decimal::from_atomics(self.amount.amount, 0).unwrap()
+    pub fn dec_amount(&self) -> StdResult<Decimal> {
+        self.amount.amount.into_base_decimal()
     }
 
     pub fn proxy_storage_key(&self) -> OwnerProxySubKey {

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

@@ -1,7 +1,7 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use cosmwasm_std::Decimal;
+use cosmwasm_std::{Decimal, StdError, StdResult, Uint128};
 
 pub fn compare_decimals(a: Decimal, b: Decimal, epsilon: Option<Decimal>) {
     let epsilon = epsilon.unwrap_or_else(|| Decimal::from_ratio(1u128, 100_000_000u128));
@@ -11,3 +11,23 @@ pub fn compare_decimals(a: Decimal, b: Decimal, epsilon: Option<Decimal>) {
         assert!(b - a < epsilon, "{} != {}", a, b)
     }
 }
+
+pub fn into_base_decimal(val: impl Into<Uint128>) -> StdResult<Decimal> {
+    val.into_base_decimal()
+}
+
+pub trait IntoBaseDecimal {
+    fn into_base_decimal(self) -> StdResult<Decimal>;
+}
+
+impl<T> IntoBaseDecimal for T
+where
+    T: Into<Uint128>,
+{
+    fn into_base_decimal(self) -> StdResult<Decimal> {
+        let atomics = self.into();
+        Decimal::from_atomics(atomics, 0).map_err(|_| StdError::GenericErr {
+            msg: format!("Decimal range exceeded for {atomics} with 0 decimal places."),
+        })
+    }
+}

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

@@ -145,14 +145,17 @@ impl JsonSchema for Interval {
 impl Interval {
     /// Initialize epoch in the contract with default values.
     pub fn init_interval(epochs_in_interval: u32, epoch_length: Duration, env: &Env) -> Self {
+        // if this fails it means the value provided from the chain itself (via cosmwasm) is invalid,
+        // so we really have to panic here as anything beyond that point would be invalid anyway
+        #[allow(clippy::expect_used)]
+        let current_epoch_start =
+            OffsetDateTime::from_unix_timestamp(env.block.time.seconds() as i64)
+                .expect("The timestamp provided via env.block.time is invalid");
+
         Interval {
             id: 0,
             epochs_in_interval,
-            // I really don't see a way for this to fail, unless the blockchain is lying to us
-            current_epoch_start: OffsetDateTime::from_unix_timestamp(
-                env.block.time.seconds() as i64
-            )
-            .expect("Invalid timestamp from env.block.time"),
+            current_epoch_start,
             current_epoch_id: 0,
             epoch_length,
             total_elapsed_epochs: 0,

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

@@ -1,6 +1,9 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+#![warn(clippy::expect_used)]
+#![warn(clippy::unwrap_used)]
+
 mod constants;
 pub mod delegation;
 pub mod error;

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

@@ -4,13 +4,14 @@
 // due to code generated by JsonSchema
 #![allow(clippy::field_reassign_with_default)]
 
-use crate::constants::UNIT_DELEGATION_BASE;
+use crate::constants::{TOKEN_SUPPLY, UNIT_DELEGATION_BASE};
 use crate::error::MixnetContractError;
+use crate::helpers::IntoBaseDecimal;
 use crate::reward_params::{NodeRewardParams, RewardingParams};
 use crate::rewarding::helpers::truncate_reward;
 use crate::rewarding::RewardDistribution;
 use crate::{Delegation, EpochId, IdentityKey, MixId, Percent, SphinxKey};
-use cosmwasm_std::{Addr, Coin, Decimal, Uint128};
+use cosmwasm_std::{Addr, Coin, Decimal, StdResult, Uint128};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
@@ -64,7 +65,7 @@ impl MixNodeDetails {
         self.rewarding_details.pending_operator_reward(pledge)
     }
 
-    pub fn pending_detailed_operator_reward(&self) -> Decimal {
+    pub fn pending_detailed_operator_reward(&self) -> StdResult<Decimal> {
         let pledge = self.original_pledge();
         self.rewarding_details
             .pending_detailed_operator_reward(pledge)
@@ -107,16 +108,21 @@ impl MixNodeRewarding {
         cost_params: MixNodeCostParams,
         initial_pledge: &Coin,
         current_epoch: EpochId,
-    ) -> Self {
-        MixNodeRewarding {
+    ) -> Result<Self, MixnetContractError> {
+        assert!(
+            initial_pledge.amount <= TOKEN_SUPPLY,
+            "pledge cannot be larger than the token supply"
+        );
+
+        Ok(MixNodeRewarding {
             cost_params,
-            operator: Decimal::from_atomics(initial_pledge.amount, 0).unwrap(),
+            operator: initial_pledge.amount.into_base_decimal()?,
             delegates: Decimal::zero(),
             total_unit_reward: Decimal::zero(),
             unit_delegation: UNIT_DELEGATION_BASE,
             last_rewarded_epoch: current_epoch,
             unique_delegations: 0,
-        }
+        })
     }
 
     /// Determines whether this node is still bonded. This is performed via a simple check,
@@ -135,27 +141,30 @@ impl MixNodeRewarding {
         }
     }
 
-    pub fn pending_detailed_operator_reward(&self, original_pledge: &Coin) -> Decimal {
-        let initial_dec = Decimal::from_atomics(original_pledge.amount, 0).unwrap();
+    pub fn pending_detailed_operator_reward(&self, original_pledge: &Coin) -> StdResult<Decimal> {
+        let initial_dec = original_pledge.amount.into_base_decimal()?;
         if initial_dec > self.operator {
             panic!(
                 "seems slashing has occurred while it has not been implemented nor accounted for!"
             )
         }
-        self.operator - initial_dec
+        Ok(self.operator - initial_dec)
     }
 
     pub fn operator_pledge_with_reward(&self, denom: impl Into<String>) -> Coin {
         truncate_reward(self.operator, denom)
     }
 
-    pub fn pending_delegator_reward(&self, delegation: &Delegation) -> Coin {
-        let delegator_reward = self.determine_delegation_reward(delegation);
-        truncate_reward(delegator_reward, &delegation.amount.denom)
+    pub fn pending_delegator_reward(&self, delegation: &Delegation) -> StdResult<Coin> {
+        let delegator_reward = self.determine_delegation_reward(delegation)?;
+        Ok(truncate_reward(delegator_reward, &delegation.amount.denom))
     }
 
-    pub fn withdraw_operator_reward(&mut self, original_pledge: &Coin) -> Coin {
-        let initial_dec = Decimal::from_atomics(original_pledge.amount, 0).unwrap();
+    pub fn withdraw_operator_reward(
+        &mut self,
+        original_pledge: &Coin,
+    ) -> Result<Coin, MixnetContractError> {
+        let initial_dec = original_pledge.amount.into_base_decimal()?;
         if initial_dec > self.operator {
             panic!(
                 "seems slashing has occurred while it has not been implemented nor accounted for!"
@@ -164,14 +173,14 @@ impl MixNodeRewarding {
         let diff = self.operator - initial_dec;
         self.operator = initial_dec;
 
-        truncate_reward(diff, &original_pledge.denom)
+        Ok(truncate_reward(diff, &original_pledge.denom))
     }
 
     pub fn withdraw_delegator_reward(
         &mut self,
         delegation: &mut Delegation,
     ) -> Result<Coin, MixnetContractError> {
-        let reward = self.determine_delegation_reward(delegation);
+        let reward = self.determine_delegation_reward(delegation)?;
         self.decrease_delegates_decimal(reward)?;
 
         delegation.cumulative_reward_ratio = self.full_reward_ratio();
@@ -301,23 +310,27 @@ impl MixNodeRewarding {
         self.distribute_rewards(reward_distribution, absolute_epoch_id)
     }
 
-    pub fn determine_delegation_reward(&self, delegation: &Delegation) -> Decimal {
+    pub fn determine_delegation_reward(&self, delegation: &Delegation) -> StdResult<Decimal> {
         let starting_ratio = delegation.cumulative_reward_ratio;
         let ending_ratio = self.full_reward_ratio();
         let adjust = starting_ratio + self.unit_delegation;
 
-        (ending_ratio - starting_ratio) * delegation.dec_amount() / adjust
+        Ok((ending_ratio - starting_ratio) * delegation.dec_amount()? / adjust)
     }
 
     // this updates `unique_delegations` field
-    pub fn add_base_delegation(&mut self, amount: Uint128) {
-        self.increase_delegates_uint128(amount);
+    pub fn add_base_delegation(&mut self, amount: Uint128) -> Result<(), MixnetContractError> {
+        self.increase_delegates_uint128(amount)?;
         self.unique_delegations += 1;
+        Ok(())
     }
 
-    pub fn increase_delegates_uint128(&mut self, amount: Uint128) {
-        // the unwrap here is fine as the value is guaranteed to fit under provided constraints
-        self.delegates += Decimal::from_atomics(amount, 0).unwrap()
+    pub fn increase_delegates_uint128(
+        &mut self,
+        amount: Uint128,
+    ) -> Result<(), MixnetContractError> {
+        self.delegates += amount.into_base_decimal()?;
+        Ok(())
     }
 
     // this updates `unique_delegations` field
@@ -335,7 +348,7 @@ impl MixNodeRewarding {
         &mut self,
         amount: Uint128,
     ) -> Result<(), MixnetContractError> {
-        let amount_dec = Decimal::from_atomics(amount, 0).unwrap();
+        let amount_dec = amount.into_base_decimal()?;
         self.decrease_delegates_decimal(amount_dec)
     }
 
@@ -368,8 +381,8 @@ impl MixNodeRewarding {
     }
 
     pub fn undelegate(&mut self, delegation: &Delegation) -> Result<Coin, MixnetContractError> {
-        let reward = self.determine_delegation_reward(delegation);
-        let full_amount = reward + delegation.dec_amount();
+        let reward = self.determine_delegation_reward(delegation)?;
+        let full_amount = reward + delegation.dec_amount()?;
         self.remove_delegation_decimal(full_amount)?;
         Ok(truncate_reward(full_amount, &delegation.amount.denom))
     }

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

@@ -2,6 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::delegation::OwnerProxySubKey;
+use crate::error::MixnetContractError;
+use crate::helpers::IntoBaseDecimal;
 use crate::mixnode::{MixNodeConfigUpdate, MixNodeCostParams};
 use crate::reward_params::{
     IntervalRewardParams, IntervalRewardingParamsUpdate, Performance, RewardingParams,
@@ -41,14 +43,17 @@ pub struct InitialRewardingParams {
 }
 
 impl InitialRewardingParams {
-    pub fn into_rewarding_params(self, epochs_in_interval: u32) -> RewardingParams {
+    pub fn into_rewarding_params(
+        self,
+        epochs_in_interval: u32,
+    ) -> Result<RewardingParams, MixnetContractError> {
         let epoch_reward_budget = self.initial_reward_pool
-            / Decimal::from_atomics(epochs_in_interval, 0).unwrap()
+            / epochs_in_interval.into_base_decimal()?
             * self.interval_pool_emission;
         let stake_saturation_point =
-            self.initial_staking_supply / Decimal::from_atomics(self.rewarded_set_size, 0).unwrap();
+            self.initial_staking_supply / self.rewarded_set_size.into_base_decimal()?;
 
-        RewardingParams {
+        Ok(RewardingParams {
             interval: IntervalRewardParams {
                 reward_pool: self.initial_reward_pool,
                 staking_supply: self.initial_staking_supply,
@@ -61,7 +66,7 @@ impl InitialRewardingParams {
             },
             rewarded_set_size: self.rewarded_set_size,
             active_set_size: self.active_set_size,
-        }
+        })
     }
 }
 
@@ -408,4 +413,6 @@ pub enum QueryMsg {
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
 #[serde(rename_all = "snake_case")]
-pub struct MigrateMsg {}
+pub struct MigrateMsg {
+    pub vesting_contract_address: Option<String>,
+}

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

@@ -1,6 +1,7 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::helpers::IntoBaseDecimal;
 use crate::{error::MixnetContractError, Percent};
 use cosmwasm_std::Decimal;
 use schemars::JsonSchema;
@@ -110,24 +111,33 @@ impl RewardingParams {
     pub fn dec_rewarded_set_size(&self) -> Decimal {
         // the unwrap here is fine as we're guaranteed an `u32` is going to fit in a Decimal
         // with 0 decimal places
-        Decimal::from_atomics(self.rewarded_set_size, 0).unwrap()
+        #[allow(clippy::unwrap_used)]
+        self.rewarded_set_size.into_base_decimal().unwrap()
     }
 
     pub fn dec_active_set_size(&self) -> Decimal {
         // the unwrap here is fine as we're guaranteed an `u32` is going to fit in a Decimal
         // with 0 decimal places
-        Decimal::from_atomics(self.active_set_size, 0).unwrap()
+        #[allow(clippy::unwrap_used)]
+        self.active_set_size.into_base_decimal().unwrap()
     }
 
     fn dec_standby_set_size(&self) -> Decimal {
         // the unwrap here is fine as we're guaranteed an `u32` is going to fit in a Decimal
         // with 0 decimal places
-        Decimal::from_atomics(self.rewarded_set_size - self.active_set_size, 0).unwrap()
+        #[allow(clippy::unwrap_used)]
+        (self.rewarded_set_size - self.active_set_size)
+            .into_base_decimal()
+            .unwrap()
     }
 
     pub fn apply_epochs_in_interval_change(&mut self, new_epochs_in_interval: u32) {
-        self.interval.epoch_reward_budget = self.interval.reward_pool
-            / Decimal::from_atomics(new_epochs_in_interval, 0).unwrap()
+        // the unwrap here is fine as we're guaranteed an `u32` is going to fit in a Decimal
+        // with 0 decimal places
+        #[allow(clippy::unwrap_used)]
+        let new_epochs_in_interval = new_epochs_in_interval.into_base_decimal().unwrap();
+
+        self.interval.epoch_reward_budget = self.interval.reward_pool / new_epochs_in_interval
             * self.interval.interval_pool_emission;
     }
 
@@ -200,13 +210,13 @@ impl RewardingParams {
 
         if recompute_epoch_budget {
             self.interval.epoch_reward_budget = self.interval.reward_pool
-                / Decimal::from_atomics(epochs_in_interval, 0).unwrap()
+                / epochs_in_interval.into_base_decimal()?
                 * self.interval.interval_pool_emission;
         }
 
         if recompute_saturation_point {
-            self.interval.stake_saturation_point = self.interval.staking_supply
-                / Decimal::from_atomics(self.rewarded_set_size, 0).unwrap();
+            self.interval.stake_saturation_point =
+                self.interval.staking_supply / self.rewarded_set_size.into_base_decimal()?
         }
 
         Ok(())

common/cosmwasm-smart-contracts/mixnet-contract/src/rewarding/mod.rs

@@ -29,7 +29,7 @@ pub struct RewardEstimate {
     pub operating_cost: Decimal,
 }
 
-#[derive(Clone, Copy, Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, JsonSchema, PartialEq, Eq)]
 pub struct RewardDistribution {
     pub operator: Decimal,
     pub delegates: Decimal,

common/cosmwasm-smart-contracts/mixnet-contract/src/rewarding/simulator/mod.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::MixnetContractError;
+use crate::helpers::IntoBaseDecimal;
 use crate::reward_params::NodeRewardParams;
 use crate::rewarding::simulator::simulated_node::SimulatedNode;
 use crate::rewarding::RewardDistribution;
@@ -33,7 +34,7 @@ impl Simulator {
         }
     }
 
-    fn advance_epoch(&mut self) {
+    fn advance_epoch(&mut self) -> Result<(), MixnetContractError> {
         let updated = self.interval.advance_epoch();
 
         // we rolled over an interval
@@ -47,10 +48,13 @@ impl Simulator {
                     .staking_supply_scale_factor
                     * self.pending_reward_pool_emission;
             let epoch_reward_budget = reward_pool
-                / Decimal::from_atomics(self.interval.epochs_in_interval(), 0).unwrap()
+                / self.interval.epochs_in_interval().into_base_decimal()?
                 * old.interval_pool_emission.value();
             let stake_saturation_point = staking_supply
-                / Decimal::from_atomics(self.system_rewarding_params.rewarded_set_size, 0).unwrap();
+                / self
+                    .system_rewarding_params
+                    .rewarded_set_size
+                    .into_base_decimal()?;
 
             let updated_params = RewardingParams {
                 interval: IntervalRewardParams {
@@ -71,9 +75,15 @@ impl Simulator {
             self.pending_reward_pool_emission = Decimal::zero();
         }
         self.interval = updated;
+
+        Ok(())
     }
 
-    pub fn bond(&mut self, pledge: Coin, cost_params: MixNodeCostParams) -> MixId {
+    pub fn bond(
+        &mut self,
+        pledge: Coin,
+        cost_params: MixNodeCostParams,
+    ) -> Result<MixId, MixnetContractError> {
         let mix_id = self.next_mix_id;
 
         self.nodes.insert(
@@ -83,16 +93,24 @@ impl Simulator {
                 cost_params,
                 &pledge,
                 self.interval.current_epoch_absolute_id(),
-            ),
+            )?,
         );
 
         self.next_mix_id += 1;
 
-        mix_id
+        Ok(mix_id)
     }
 
-    pub fn delegate<S: Into<String>>(&mut self, delegator: S, delegation: Coin, mix_id: MixId) {
-        let node = self.nodes.get_mut(&mix_id).expect("node doesn't exist");
+    pub fn delegate<S: Into<String>>(
+        &mut self,
+        delegator: S,
+        delegation: Coin,
+        mix_id: MixId,
+    ) -> Result<(), MixnetContractError> {
+        let node = self
+            .nodes
+            .get_mut(&mix_id)
+            .ok_or(MixnetContractError::MixNodeBondNotFound { mix_id })?;
         node.delegate(delegator, delegation)
     }
 
@@ -103,23 +121,35 @@ impl Simulator {
         delegator: S,
         mix_id: MixId,
     ) -> Result<(Coin, Coin), MixnetContractError> {
-        let node = self.nodes.get_mut(&mix_id).expect("node not found");
+        let node = self
+            .nodes
+            .get_mut(&mix_id)
+            .ok_or(MixnetContractError::MixNodeBondNotFound { mix_id })?;
         node.undelegate(delegator)
     }
 
-    pub fn simulate_epoch_single_node(&mut self, params: NodeRewardParams) -> RewardDistribution {
+    pub fn simulate_epoch_single_node(
+        &mut self,
+        params: NodeRewardParams,
+    ) -> Result<RewardDistribution, MixnetContractError> {
         assert_eq!(self.nodes.len(), 1);
 
-        let id = *self.nodes.keys().next().unwrap();
-        let mut params_map = BTreeMap::new();
-        params_map.insert(id, params);
-        self.simulate_epoch(&params_map).remove(&id).unwrap()
+        if let Some(&id) = self.nodes.keys().next() {
+            let mut params_map = BTreeMap::new();
+            params_map.insert(id, params);
+            Ok(self
+                .simulate_epoch(&params_map)?
+                .remove(&id)
+                .unwrap_or_default())
+        } else {
+            Ok(RewardDistribution::default())
+        }
     }
 
     pub fn simulate_epoch(
         &mut self,
         node_params: &BTreeMap<MixId, NodeRewardParams>,
-    ) -> BTreeMap<MixId, RewardDistribution> {
+    ) -> Result<BTreeMap<MixId, RewardDistribution>, MixnetContractError> {
         let mut params_keys = node_params.keys().copied().collect::<Vec<_>>();
         params_keys.sort_unstable();
         let mut node_keys = self.nodes.keys().copied().collect::<Vec<_>>();
@@ -147,34 +177,41 @@ impl Simulator {
             dist.insert(*mix_id, reward_distribution);
         }
 
-        self.advance_epoch();
-        dist
+        self.advance_epoch()?;
+        Ok(dist)
     }
 
-    pub fn determine_delegation_reward(&self, delegation: &Delegation) -> Decimal {
-        self.nodes[&delegation.mix_id]
+    pub fn determine_delegation_reward(
+        &self,
+        delegation: &Delegation,
+    ) -> Result<Decimal, MixnetContractError> {
+        Ok(self.nodes[&delegation.mix_id]
             .rewarding_details
-            .determine_delegation_reward(delegation)
+            .determine_delegation_reward(delegation)?)
     }
 
-    pub fn determine_total_delegation_reward(&self) -> Decimal {
+    pub fn determine_total_delegation_reward(&self) -> Result<Decimal, MixnetContractError> {
         let mut total = Decimal::zero();
 
         for node in self.nodes.values() {
             for delegation in node.delegations.values() {
                 total += node
                     .rewarding_details
-                    .determine_delegation_reward(delegation)
+                    .determine_delegation_reward(delegation)?
             }
         }
-        total
+        Ok(total)
     }
 
     // assume node state doesn't change in the interval (kinda unrealistic)
-    pub fn simulate_full_interval(&mut self, node_params: &BTreeMap<MixId, NodeRewardParams>) {
+    pub fn simulate_full_interval(
+        &mut self,
+        node_params: &BTreeMap<MixId, NodeRewardParams>,
+    ) -> Result<(), MixnetContractError> {
         for _ in 0..self.interval.epochs_in_interval() {
-            self.simulate_epoch(node_params);
+            self.simulate_epoch(node_params)?;
         }
+        Ok(())
     }
 }
 
@@ -197,6 +234,10 @@ mod tests {
         use cosmwasm_std::testing::mock_env;
         use std::time::Duration;
 
+        // explicitly marking this as part of #[allow(clippy::unwrap_used)] until
+        // https://github.com/rust-lang/rust-clippy/pull/9686
+        // is merged into a release
+        #[allow(clippy::unwrap_used)]
         fn base_simulator(initial_pledge: u128) -> Simulator {
             let profit_margin = Percent::from_percentage_value(10).unwrap();
             let interval_operating_cost = Coin::new(40_000_000, "unym");
@@ -238,20 +279,32 @@ mod tests {
                 profit_margin_percent: profit_margin,
                 interval_operating_cost,
             };
-            simulator.bond(initial_pledge, cost_params);
+            simulator.bond(initial_pledge, cost_params).unwrap();
             simulator
         }
 
         // essentially our delegations + estimated rewards HAVE TO equal to what we actually determined
+        //
+        // explicitly marking this as part of #[allow(clippy::unwrap_used)] until
+        // https://github.com/rust-lang/rust-clippy/pull/9686
+        // is merged into a release
+        #[allow(clippy::unwrap_used)]
         fn check_rewarding_invariant(simulator: &Simulator) {
             for node in simulator.nodes.values() {
-                let delegation_sum: Decimal =
-                    node.delegations.values().map(|d| d.dec_amount()).sum();
+                let delegation_sum: Decimal = node
+                    .delegations
+                    .values()
+                    .map(|d| d.dec_amount().unwrap())
+                    .sum();
 
                 let reward_sum: Decimal = node
                     .delegations
                     .values()
-                    .map(|d| node.rewarding_details.determine_delegation_reward(d))
+                    .map(|d| {
+                        node.rewarding_details
+                            .determine_delegation_reward(d)
+                            .unwrap()
+                    })
                     .sum();
 
                 // let reward_sum = simulator.determine_total_delegation_reward();
@@ -269,7 +322,7 @@ mod tests {
 
             let epoch_params =
                 NodeRewardParams::new(Percent::from_percentage_value(100).unwrap(), true);
-            let rewards = simulator.simulate_epoch_single_node(epoch_params);
+            let rewards = simulator.simulate_epoch_single_node(epoch_params).unwrap();
 
             assert_eq!(rewards.delegates, Decimal::zero());
             compare_decimals(
@@ -282,11 +335,13 @@ mod tests {
         #[test]
         fn single_delegation_at_genesis() {
             let mut simulator = base_simulator(10000_000000);
-            simulator.delegate("alice", Coin::new(18000_000000, "unym"), 0);
+            simulator
+                .delegate("alice", Coin::new(18000_000000, "unym"), 0)
+                .unwrap();
 
             let node_params =
                 NodeRewardParams::new(Percent::from_percentage_value(100).unwrap(), true);
-            let rewards = simulator.simulate_epoch_single_node(node_params);
+            let rewards = simulator.simulate_epoch_single_node(node_params).unwrap();
 
             compare_decimals(
                 rewards.delegates,
@@ -297,7 +352,7 @@ mod tests {
 
             compare_decimals(
                 rewards.delegates,
-                simulator.determine_total_delegation_reward(),
+                simulator.determine_total_delegation_reward().unwrap(),
                 None,
             );
             let node = &simulator.nodes[&0];
@@ -317,20 +372,22 @@ mod tests {
             let node_params =
                 NodeRewardParams::new(Percent::from_percentage_value(100).unwrap(), true);
 
-            let rewards1 = simulator.simulate_epoch_single_node(node_params);
+            let rewards1 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator1 = "1128452.5416104363".parse().unwrap();
             assert_eq!(rewards1.delegates, Decimal::zero());
             compare_decimals(rewards1.operator, expected_operator1, None);
 
-            simulator.delegate("alice", Coin::new(18000_000000, "unym"), 0);
+            simulator
+                .delegate("alice", Coin::new(18000_000000, "unym"), 0)
+                .unwrap();
 
-            let rewards2 = simulator.simulate_epoch_single_node(node_params);
+            let rewards2 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator2 = "1363843.413584609".parse().unwrap();
             let expected_delegator_reward1 = "1795952.25874404".parse().unwrap();
             compare_decimals(rewards2.delegates, expected_delegator_reward1, None);
             compare_decimals(rewards2.operator, expected_operator2, None);
 
-            let rewards3 = simulator.simulate_epoch_single_node(node_params);
+            let rewards3 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator3 = "1364017.7824440491".parse().unwrap();
             let expected_delegator_reward2 = "1796135.9269468693".parse().unwrap();
             compare_decimals(rewards3.delegates, expected_delegator_reward2, None);
@@ -364,11 +421,15 @@ mod tests {
 
             // add 2 delegations at genesis (because it makes things easier and as shown with previous tests
             // delegating at different times still work)
-            simulator.delegate("alice", Coin::new(18000_000000, "unym"), 0);
-            simulator.delegate("bob", Coin::new(4000_000000, "unym"), 0);
+            simulator
+                .delegate("alice", Coin::new(18000_000000, "unym"), 0)
+                .unwrap();
+            simulator
+                .delegate("bob", Coin::new(4000_000000, "unym"), 0)
+                .unwrap();
 
             // "normal", sanity check rewarding
-            let rewards1 = simulator.simulate_epoch_single_node(node_params);
+            let rewards1 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator1 = "1411087.1007647323".parse().unwrap();
             let expected_delegator_reward1 = "2199961.032388664".parse().unwrap();
             compare_decimals(rewards1.delegates, expected_delegator_reward1, None);
@@ -378,14 +439,15 @@ mod tests {
             let node = simulator.nodes.get_mut(&0).unwrap();
             let reward = node
                 .rewarding_details
-                .withdraw_operator_reward(&original_pledge);
+                .withdraw_operator_reward(&original_pledge)
+                .unwrap();
             assert_eq!(reward.amount, truncate_reward_amount(expected_operator1));
             assert_eq!(
                 node.rewarding_details.operator,
                 Decimal::from_atomics(original_pledge.amount, 0).unwrap()
             );
 
-            let rewards2 = simulator.simulate_epoch_single_node(node_params);
+            let rewards2 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator2 = "1411113.0004067947".parse().unwrap();
             let expected_delegator_reward2 = "2200183.3879084454".parse().unwrap();
             compare_decimals(rewards2.delegates, expected_delegator_reward2, None);
@@ -402,11 +464,15 @@ mod tests {
 
             // add 2 delegations at genesis (because it makes things easier and as shown with previous tests
             // delegating at different times still work)
-            simulator.delegate("alice", Coin::new(18000_000000, "unym"), 0);
-            simulator.delegate("bob", Coin::new(4000_000000, "unym"), 0);
+            simulator
+                .delegate("alice", Coin::new(18000_000000, "unym"), 0)
+                .unwrap();
+            simulator
+                .delegate("bob", Coin::new(4000_000000, "unym"), 0)
+                .unwrap();
 
             // "normal", sanity check rewarding
-            let rewards1 = simulator.simulate_epoch_single_node(node_params);
+            let rewards1 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator1 = "1411087.1007647323".parse().unwrap();
             let expected_delegator_reward1 = "2199961.032388664".parse().unwrap();
             compare_decimals(rewards1.delegates, expected_delegator_reward1, None);
@@ -424,7 +490,7 @@ mod tests {
             assert_eq!(reward.amount, truncate_reward_amount(expected_del1_reward));
 
             // new reward after withdrawal
-            let rewards2 = simulator.simulate_epoch_single_node(node_params);
+            let rewards2 = simulator.simulate_epoch_single_node(node_params).unwrap();
             let expected_operator2 = "1411250.1907492676".parse().unwrap();
             let expected_delegator_reward2 = "2200004.051009689".parse().unwrap();
             compare_decimals(rewards2.delegates, expected_delegator_reward2, None);
@@ -467,22 +533,30 @@ mod tests {
             let mut performance = Percent::from_percentage_value(100).unwrap();
             for epoch in 0..720 {
                 if epoch == 0 {
-                    simulator.delegate("a", Coin::new(18000_000000, "unym"), 0)
+                    simulator
+                        .delegate("a", Coin::new(18000_000000, "unym"), 0)
+                        .unwrap()
                 }
                 if epoch == 42 {
-                    simulator.delegate("b", Coin::new(2000_000000, "unym"), 0)
+                    simulator
+                        .delegate("b", Coin::new(2000_000000, "unym"), 0)
+                        .unwrap()
                 }
                 if epoch == 89 {
                     is_active = false;
                 }
                 if epoch == 123 {
-                    simulator.delegate("c", Coin::new(6666_000000, "unym"), 0)
+                    simulator
+                        .delegate("c", Coin::new(6666_000000, "unym"), 0)
+                        .unwrap()
                 }
                 if epoch == 167 {
                     performance = Percent::from_percentage_value(90).unwrap();
                 }
                 if epoch == 245 {
-                    simulator.delegate("d", Coin::new(2050_000000, "unym"), 0)
+                    simulator
+                        .delegate("d", Coin::new(2050_000000, "unym"), 0)
+                        .unwrap()
                 }
                 if epoch == 264 {
                     let (delegation, _reward) = simulator.undelegate("b", 0).unwrap();
@@ -503,13 +577,15 @@ mod tests {
                     // TODO: figure out if there's a good way to verify whether `reward` is what we expect it to be
                 }
                 if epoch == 545 {
-                    simulator.delegate("e", Coin::new(5000_000000, "unym"), 0)
+                    simulator
+                        .delegate("e", Coin::new(5000_000000, "unym"), 0)
+                        .unwrap()
                 }
 
                 // this has to always hold
                 check_rewarding_invariant(&simulator);
                 let node_params = NodeRewardParams::new(performance, is_active);
-                simulator.simulate_epoch_single_node(node_params);
+                simulator.simulate_epoch_single_node(node_params).unwrap();
             }
 
             // after everyone undelegates, there should be nothing left in the delegates pool
@@ -563,95 +639,135 @@ mod tests {
 
         let mut simulator = Simulator::new(rewarding_params, interval);
 
-        let n0 = simulator.bond(
-            Coin::new(11_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(1_000_000_000000, "unym"), n0);
+        let n0 = simulator
+            .bond(
+                Coin::new(11_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(1_000_000_000000, "unym"), n0)
+            .unwrap();
 
-        let n1 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(11_000_000_000000, "unym"), n1);
+        let n1 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(11_000_000_000000, "unym"), n1)
+            .unwrap();
 
-        let n2 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(9_000_000_000000, "unym"), n2);
+        let n2 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(9_000_000_000000, "unym"), n2)
+            .unwrap();
 
-        let n3 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(0).unwrap(),
-                interval_operating_cost: Coin::new(500_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(7_000_000_000000, "unym"), n3);
+        let n3 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(0).unwrap(),
+                    interval_operating_cost: Coin::new(500_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(7_000_000_000000, "unym"), n3)
+            .unwrap();
 
-        let n4 = simulator.bond(
-            Coin::new(1000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(7_999_000_000000, "unym"), n4);
+        let n4 = simulator
+            .bond(
+                Coin::new(1000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(7_999_000_000000, "unym"), n4)
+            .unwrap();
 
-        let n5 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(7_000_000_000000, "unym"), n5);
+        let n5 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(7_000_000_000000, "unym"), n5)
+            .unwrap();
 
-        let n6 = simulator.bond(
-            Coin::new(11_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(1_000_000_000000, "unym"), n6);
+        let n6 = simulator
+            .bond(
+                Coin::new(11_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(1_000_000_000000, "unym"), n6)
+            .unwrap();
 
-        let n7 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(9_000_000_000000, "unym"), n7);
+        let n7 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(9_000_000_000000, "unym"), n7)
+            .unwrap();
 
-        let n8 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(0).unwrap(),
-                interval_operating_cost: Coin::new(500_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(7_000_000_000000, "unym"), n8);
+        let n8 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(0).unwrap(),
+                    interval_operating_cost: Coin::new(500_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(7_000_000_000000, "unym"), n8)
+            .unwrap();
 
-        let n9 = simulator.bond(
-            Coin::new(1_000_000_000000, "unym"),
-            MixNodeCostParams {
-                profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
-                interval_operating_cost: Coin::new(40_000_000, "unym"),
-            },
-        );
-        simulator.delegate("delegator", Coin::new(7_000_000_000000, "unym"), n9);
+        let n9 = simulator
+            .bond(
+                Coin::new(1_000_000_000000, "unym"),
+                MixNodeCostParams {
+                    profit_margin_percent: Percent::from_percentage_value(10).unwrap(),
+                    interval_operating_cost: Coin::new(40_000_000, "unym"),
+                },
+            )
+            .unwrap();
+        simulator
+            .delegate("delegator", Coin::new(7_000_000_000000, "unym"), n9)
+            .unwrap();
 
         let uptime_1 = Percent::from_percentage_value(100).unwrap();
         let uptime_09 = Percent::from_percentage_value(90).unwrap();
@@ -673,7 +789,7 @@ mod tests {
         .collect::<BTreeMap<_, _>>();
 
         for _ in 0..23 {
-            simulator.simulate_full_interval(&node_params);
+            simulator.simulate_full_interval(&node_params).unwrap();
         }
 
         // we allow the delta to be within 0.1unym,

common/cosmwasm-smart-contracts/mixnet-contract/src/rewarding/simulator/simulated_node.rs

@@ -20,21 +20,25 @@ impl SimulatedNode {
         cost_params: MixNodeCostParams,
         initial_pledge: &Coin,
         current_epoch: EpochId,
-    ) -> Self {
-        SimulatedNode {
+    ) -> Result<Self, MixnetContractError> {
+        Ok(SimulatedNode {
             mix_id,
             rewarding_details: MixNodeRewarding::initialise_new(
                 cost_params,
                 initial_pledge,
                 current_epoch,
-            ),
+            )?,
             delegations: HashMap::new(),
-        }
+        })
     }
 
-    pub fn delegate<S: Into<String>>(&mut self, delegator: S, delegation: Coin) {
+    pub fn delegate<S: Into<String>>(
+        &mut self,
+        delegator: S,
+        delegation: Coin,
+    ) -> Result<(), MixnetContractError> {
         self.rewarding_details
-            .add_base_delegation(delegation.amount);
+            .add_base_delegation(delegation.amount)?;
 
         let delegator = delegator.into();
         let delegation = Delegation::new(
@@ -47,6 +51,7 @@ impl SimulatedNode {
         );
 
         self.delegations.insert(delegator, delegation);
+        Ok(())
     }
 
     pub fn undelegate<S: Into<String>>(
@@ -54,16 +59,19 @@ impl SimulatedNode {
         delegator: S,
     ) -> Result<(Coin, Coin), MixnetContractError> {
         let delegator = delegator.into();
-        let delegation = self
-            .delegations
-            .remove(&delegator)
-            .expect("delegation not found");
+        let delegation = self.delegations.remove(&delegator).ok_or(
+            MixnetContractError::NoMixnodeDelegationFound {
+                mix_id: MixId::MAX,
+                address: delegator,
+                proxy: None,
+            },
+        )?;
 
         let reward = self
             .rewarding_details
-            .determine_delegation_reward(&delegation);
+            .determine_delegation_reward(&delegation)?;
         self.rewarding_details
-            .remove_delegation_decimal(delegation.dec_amount() + reward)?;
+            .remove_delegation_decimal(delegation.dec_amount()? + reward)?;
 
         let reward_denom = &delegation.amount.denom;
         let truncated_reward = truncate_reward(reward, reward_denom);

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

@@ -35,6 +35,10 @@ impl LayerDistribution {
             (Layer::Two, self.layer2),
             (Layer::Three, self.layer3),
         ];
+
+        // we explicitly put 3 elements into the iterator, so the iterator is DEFINITELY
+        // not empty and thus the unwrap cannot fail
+        #[allow(clippy::unwrap_used)]
         layers.iter().min_by_key(|x| x.1).unwrap().0
     }
 

common/cosmwasm-smart-contracts/vesting-contract/Cargo.toml

@@ -9,7 +9,6 @@ mixnet-contract-common = { path = "../mixnet-contract" }
 contracts-common = { path = "../contracts-common" }
 serde = { version = "1.0", features = ["derive"] }
 schemars = "0.8"
-log = "0.4"
 ts-rs = {version = "6.1.2", optional = true}
 
 [features]

common/cosmwasm-smart-contracts/vesting-contract/src/lib.rs

@@ -1,13 +1,15 @@
-use std::str::FromStr;
-
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
+
+#![warn(clippy::expect_used)]
+#![warn(clippy::unwrap_used)]
+
 use contracts_common::Percent;
 use cosmwasm_std::{Addr, Coin, Timestamp, Uint128};
-use log::warn;
 use mixnet_contract_common::MixId;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use std::str::FromStr;
 
 pub use messages::{ExecuteMsg, InitMsg, MigrateMsg, QueryMsg};
 
@@ -59,30 +61,19 @@ impl FromStr for PledgeCap {
     fn from_str(cap: &str) -> Result<Self, Self::Err> {
         let cap = cap.replace('_', "").replace(',', ".");
         match Percent::from_str(&cap) {
-            Ok(p) => {
-                if p.is_zero() {
-                    warn!("Pledge cap set to 0%, are you sure this is right?")
-                }
-                Ok(PledgeCap::Percent(p))
-            }
-            Err(_) => {
-                match cap.parse::<u128>() {
-                    Ok(i) => {
-                        if i < 100_000_000_000 {
-                            warn!("PledgeCap set to less then 100_000 NYM, are you sure this is right?");
-                        }
-                        Ok(PledgeCap::Absolute(Uint128::from(i)))
-                    }
-                    Err(_e) => Err(format!("Could not parse {} as Percent or Uint128", cap)),
-                }
-            }
+            Ok(p) => Ok(PledgeCap::Percent(p)),
+            Err(_) => match cap.parse::<u128>() {
+                Ok(i) => Ok(PledgeCap::Absolute(Uint128::from(i))),
+                Err(_e) => Err(format!("Could not parse {} as Percent or Uint128", cap)),
+            },
         }
     }
 }
 
 impl Default for PledgeCap {
     fn default() -> Self {
-        PledgeCap::Absolute(Uint128::from(100_000_000_000u128))
+        #[allow(clippy::expect_used)]
+        PledgeCap::Percent(Percent::from_percentage_value(10).expect("This can never fail!"))
     }
 }
 

common/network-defaults/src/mainnet.rs

@@ -10,7 +10,7 @@ pub const MIX_DENOM: DenomDetails = DenomDetails::new("unym", "nym", 6);
 pub const STAKE_DENOM: DenomDetails = DenomDetails::new("unyx", "nyx", 6);
 
 pub(crate) const MIXNET_CONTRACT_ADDRESS: &str =
-    "n14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9sjyvg3g";
+    "n17srjznxl9dvzdkpwpw24gg668wc73val88a6m5ajg6ankwvz9wtst0cznr";
 pub(crate) const VESTING_CONTRACT_ADDRESS: &str =
     "n1nc5tatafv6eyq7llkr2gv50ff9e22mnf70qgjlv737ktmt4eswrq73f2nw";
 pub(crate) const BANDWIDTH_CLAIM_CONTRACT_ADDRESS: &str =

common/nymcoconut/src/scheme/keygen.rs

@@ -85,7 +85,7 @@ impl SecretKey {
     // x || ys.len() || ys
     pub fn to_bytes(&self) -> Vec<u8> {
         let ys_len = self.ys.len();
-        let mut bytes = Vec::with_capacity(8 + (ys_len + 1) as usize * 32);
+        let mut bytes = Vec::with_capacity(8 + (ys_len + 1) * 32);
 
         bytes.extend_from_slice(&self.x.to_bytes());
         bytes.extend_from_slice(&ys_len.to_le_bytes());
@@ -162,7 +162,7 @@ impl TryFrom<&[u8]> for VerificationKey {
         let mut beta_g1_end: u64 = 0;
         for i in 0..betas_len {
             let start = (104 + i * 48) as usize;
-            let end = (start + 48) as usize;
+            let end = start + 48;
             let beta_i_bytes = bytes[start..end].try_into().unwrap();
             let beta_i = try_deserialize_g1_projective(
                 &beta_i_bytes,
@@ -178,7 +178,7 @@ impl TryFrom<&[u8]> for VerificationKey {
         let mut beta_g2 = Vec::with_capacity(betas_len as usize);
         for i in 0..betas_len {
             let start = (beta_g1_end + i * 96) as usize;
-            let end = (start + 96) as usize;
+            let end = start + 96;
             let beta_i_bytes = bytes[start..end].try_into().unwrap();
             let beta_i = try_deserialize_g2_projective(
                 &beta_i_bytes,

common/nymsphinx/chunking/src/set.rs

@@ -94,10 +94,10 @@ fn prepare_unlinked_fragmented_set(
 
     for i in 1..(pre_casted_frags + 1) {
         // we can't use u8 directly here as upper (NON-INCLUSIVE, so it would always fit) bound could be u8::max_value() + 1
-        let lb = (i as usize - 1) * unlinked_fragment_payload_max_len(max_plaintext_size);
+        let lb = (i - 1) * unlinked_fragment_payload_max_len(max_plaintext_size);
         let ub = usize::min(
             message.len(),
-            i as usize * unlinked_fragment_payload_max_len(max_plaintext_size),
+            i * unlinked_fragment_payload_max_len(max_plaintext_size),
         );
         fragments.push(
             Fragment::try_new(

common/nymsphinx/params/src/packet_sizes.rs

@@ -130,6 +130,6 @@ impl PacketSize {
 
 impl Default for PacketSize {
     fn default() -> Self {
-        PacketSize::ExtendedPacket32
+        PacketSize::RegularPacket
     }
 }

common/socks5/proxy-helpers/Cargo.toml

@@ -14,8 +14,11 @@ tokio-util = { version = "0.7.3", features = [ "io" ] } # reason for getting thi
 # In the long run, the dependency should probably get removed in favour of pure-tokio implementation, but for time being it's fine.
 futures = "0.3"
 log = "0.4"
-socks5-requests = { path = "../requests" }
+
+# internal
+client-connections = { path = "../../client-connections" }
 ordered-buffer = { path = "../ordered-buffer" }
+socks5-requests = { path = "../requests" }
 task = { path = "../../task" }
 
 [dev-dependencies]

common/socks5/proxy-helpers/src/connection_controller.rs

@@ -1,6 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use client_connections::ClosedConnectionSender;
 use futures::channel::mpsc;
 use futures::StreamExt;
 use log::*;
@@ -73,6 +74,9 @@ pub struct Controller {
     // to avoid memory issues
     recently_closed: HashSet<ConnectionId>,
 
+    // Broadcast closed connections
+    closed_connection_tx: ClosedConnectionSender,
+
     // TODO: this can potentially be abused to ddos and kill provider. Not sure at this point
     // how to handle it more gracefully
 
@@ -84,13 +88,17 @@ pub struct Controller {
 }
 
 impl Controller {
-    pub fn new(shutdown: ShutdownListener) -> (Self, ControllerSender) {
+    pub fn new(
+        closed_connection_tx: ClosedConnectionSender,
+        shutdown: ShutdownListener,
+    ) -> (Self, ControllerSender) {
         let (sender, receiver) = mpsc::unbounded();
         (
             Controller {
                 active_connections: HashMap::new(),
                 receiver,
                 recently_closed: HashSet::new(),
+                closed_connection_tx,
                 pending_messages: HashMap::new(),
                 shutdown,
             },
@@ -127,6 +135,9 @@ impl Controller {
             )
         }
         self.recently_closed.insert(conn_id);
+
+        // Announce closed connections, currently used by the `OutQueueControl`.
+        self.closed_connection_tx.unbounded_send(conn_id).unwrap();
     }
 
     fn send_to_connection(&mut self, conn_id: ConnectionId, payload: Vec<u8>, is_closed: bool) {
@@ -172,11 +183,15 @@ impl Controller {
             pending.push((payload, is_closed));
         } else if !is_closed {
             error!(
-                "Tried to write to closed connection ({} bytes were 'lost)",
+                "Tried to write to closed connection {} ({} bytes were 'lost)",
+                conn_id,
                 payload.len()
             );
         } else {
-            debug!("Tried to write to closed connection, but remote is already closed")
+            debug!(
+                "Tried to write to closed connection {}, but remote is already closed",
+                conn_id
+            )
         }
     }
 

common/socks5/proxy-helpers/src/proxy_runner/inbound.rs

@@ -63,6 +63,10 @@ where
 
     // if we're sending through the mixnet increase the sequence number...
     let ordered_msg = message_sender.wrap_message(read_data.to_vec()).into_bytes();
+    log::trace!(
+        "pushing data down the input sender: size: {}",
+        ordered_msg.len()
+    );
     mix_sender
         .unbounded_send(adapter_fn(connection_id, ordered_msg, is_finished))
         .unwrap();

common/types/src/delegation.rs

@@ -62,6 +62,7 @@ pub struct DelegationWithEverything {
 
     // DEPRECATED, IF POSSIBLE TRY TO DISCONTINUE USE OF IT!
     pub pending_events: Vec<DelegationEvent>,
+    pub mixnode_is_unbonding: Option<bool>,
 }
 
 #[cfg_attr(feature = "generate-ts", derive(ts_rs::TS))]

contracts/CHANGELOG.md

@@ -1,8 +1,43 @@
 ## Unreleased
 
+## Added
+
+- Added migration code to the mixnet contract to allow updating stored vesting contract address to make it easier to deploy any future environments ([#1759],[#1769])
+
+[#1759]: https://github.com/nymtech/nym/pull/1759
+[#1769]: https://github.com/nymtech/nym/pull/1769
+
+## [nym-contracts-v1.1.0](https://github.com/nymtech/nym/tree/nym-contracts-v1.1.0) (2022-11-09)
+
+### Changed
+- mixnet-contract: rework of rewarding ([#1472]), which includes, but is not limited to:
+  - internal reward accounting was modified to be similar to the ideas presented in Cosmos' F1 paper, which results in throughput gains and no storage or gas cost bloat over time,
+  - introduced internal queues for pending epoch and interval events that only get resolved once relevant epoch/interval rolls over
+  - the contract no longer stores any historical information regarding past epochs/parameters/stake state for the purposes of rewarding
+  - a lot of queries got renamed to keep naming more consistent,
+  - introduced new utility-based queries such as a query for reward estimation for the current epoch,
+  - mixnodes are now identified by a monotonously increasing `mix_id`
+  - bonding now results in getting fresh `mix_id` and thus if given node decides to unbond and rebond, it will lose all its delegations,
+  - mixnode operators are now allowed to set their operating costs as opposed to having fixed value of 40nym/interval
+  - rewarding parameters are now correctly updated at an **interval** end
+  - rewarding parameters now include a staking supply scale factor attribute (beta in the tokenomics paper)
+  - node performance can now be more granular with internal `Decimal` representation as opposed to an `u8`
+  - node profit margin can now be more granular with internal `Decimal` representation as opposed to an `u8`
+  - mixnode operators are now allowed to change their configuration options, such as port information, without having to unbond
+  - mixnode unbonding is no longer instantaneous, instead it happens once an epoch rolls over
+  - it is now possible to query for operator and node history to see how often (and with what parameters) they rebonded
+  - other minor bugfixes and changes
+  - ...
+  - new exciting bugs to find and squash
+
+- vesting-contract: optional locked token pledge cap per account ([#1687]), defaults to 10%
+- vesting-contract: updated internal delegation storage due to mixnet contract revamp ([#1472])
+
 ### Added
 - vesting-contract: added query for obtaining contract build information ([#1726])
 
+[#1472]: https://github.com/nymtech/nym/pull/1472
+[#1687]: https://github.com/nymtech/nym/pull/1687
 [#1726]: https://github.com/nymtech/nym/pull/1726
 
 
@@ -11,12 +46,10 @@
 ### Added
 
 - vesting-contract: added queries for delegation timestamps and paged query for all vesting delegations in the contract ([#1569])
-- all binaries: added shell completion and [Fig](fig.io) spec generation ([#1638])
 
 ### Changed
 
 - mixnet-contract: compounding delegator rewards now happens instantaneously as opposed to having to wait for the current epoch to finish ([#1571])
-- network-requester: updated CLI to use `clap` macros ([#1638])
 
 ### Fixed
 
@@ -26,9 +59,8 @@
 
 [#1544]: https://github.com/nymtech/nym/pull/1544
 [#1569]: https://github.com/nymtech/nym/pull/1569
-[#1569]: https://github.com/nymtech/nym/pull/1571
+[#1571]: https://github.com/nymtech/nym/pull/1571
 [#1613]: https://github.com/nymtech/nym/pull/1613
-[#1638]: https://github.com/nymtech/nym/pull/1638
 
 ## [nym-contracts-v1.0.1](https://github.com/nymtech/nym/tree/nym-contracts-v1.0.1) (2022-06-22)
 

contracts/Cargo.lock

@@ -287,9 +287,9 @@ dependencies = [
 
 [[package]]
 name = "cpufeatures"
-version = "0.2.1"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "95059428f66df56b63431fdb4e1947ed2190586af5c5a8a8b71122bdf5a7f469"
+checksum = "28d997bd5e24a5928dd43e46dc529867e207907fe0b239c3477d924f7f2ca320"
 dependencies = [
  "libc",
 ]
@@ -930,7 +930,7 @@ checksum = "a3e378b66a060d48947b590737b30a1be76706c8dd7b8ba0f2fe3989c68a853f"
 
 [[package]]
 name = "mixnet-contract"
-version = "1.0.2"
+version = "1.1.0"
 dependencies = [
  "bs58",
  "cosmwasm-schema",
@@ -1625,7 +1625,6 @@ version = "0.1.0"
 dependencies = [
  "contracts-common",
  "cosmwasm-std",
- "log",
  "mixnet-contract-common",
  "schemars",
  "serde",

contracts/clippy.toml

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

contracts/mixnet/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mixnet-contract"
-version = "1.0.2"
+version = "1.1.0"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 edition = "2021"
 

contracts/mixnet/src/contract.rs

@@ -63,7 +63,7 @@ pub fn instantiate(
         Interval::init_interval(msg.epochs_in_interval, msg.epoch_duration, &env);
     let reward_params = msg
         .initial_rewarding_params
-        .into_rewarding_params(msg.epochs_in_interval);
+        .into_rewarding_params(msg.epochs_in_interval)?;
 
     interval_storage::initialise_storage(deps.storage, starting_interval)?;
     mixnet_params_storage::initialise_storage(deps.storage, state)?;
@@ -467,10 +467,20 @@ pub fn query(
 
 #[entry_point]
 pub fn migrate(
-    _deps: DepsMut<'_>,
+    deps: DepsMut<'_>,
     _env: Env,
-    _msg: MigrateMsg,
+    msg: MigrateMsg,
 ) -> Result<Response, MixnetContractError> {
+    // due to circular dependency on contract addresses (i.e. mixnet contract requiring vesting contract address
+    // and vesting contract requiring the mixnet contract address), if we ever want to deploy any new fresh
+    // environment, one of the contracts will HAVE TO go through a migration
+    if let Some(vesting_contract_address) = msg.vesting_contract_address {
+        let mut current_state = mixnet_params_storage::CONTRACT_STATE.load(deps.storage)?;
+        current_state.vesting_contract_address =
+            deps.api.addr_validate(&vesting_contract_address)?;
+        mixnet_params_storage::CONTRACT_STATE.save(deps.storage, &current_state)?;
+    }
+
     Ok(Default::default())
 }
 

contracts/mixnet/src/interval/pending_events.rs

@@ -107,7 +107,7 @@ pub(crate) fn delegate(
     };
 
     // add the amount we're intending to delegate (whether it's fresh or we're adding to the existing one)
-    mix_rewarding.add_base_delegation(stored_delegation_amount.amount);
+    mix_rewarding.add_base_delegation(stored_delegation_amount.amount)?;
 
     let cosmos_event = new_delegation_event(
         created_at,

contracts/mixnet/src/lib.rs

@@ -1,6 +1,9 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+#![warn(clippy::expect_used)]
+#![warn(clippy::unwrap_used)]
+
 mod constants;
 pub mod contract;
 mod delegations;

contracts/mixnet/src/mixnodes/helpers.rs

@@ -78,7 +78,7 @@ pub(crate) fn save_new_mixnode(
     let mix_id = next_mixnode_id_counter(storage)?;
     let current_epoch = interval_storage::current_interval(storage)?.current_epoch_absolute_id();
 
-    let mixnode_rewarding = MixNodeRewarding::initialise_new(cost_params, &pledge, current_epoch);
+    let mixnode_rewarding = MixNodeRewarding::initialise_new(cost_params, &pledge, current_epoch)?;
     let mixnode_bond = MixNodeBond::new(
         mix_id,
         owner,

contracts/mixnet/src/rewards/helpers.rs

@@ -4,8 +4,9 @@
 use super::storage;
 use crate::delegations::storage as delegations_storage;
 use crate::interval::storage as interval_storage;
-use cosmwasm_std::{Coin, Decimal, Storage};
+use cosmwasm_std::{Coin, Storage};
 use mixnet_contract_common::error::MixnetContractError;
+use mixnet_contract_common::helpers::IntoBaseDecimal;
 use mixnet_contract_common::mixnode::{MixNodeDetails, MixNodeRewarding};
 use mixnet_contract_common::Delegation;
 
@@ -22,11 +23,10 @@ pub(crate) fn apply_reward_pool_changes(
         + pending_pool_change.added;
     let staking_supply = rewarding_params.interval.staking_supply
         + rewarding_params.interval.staking_supply_scale_factor * pending_pool_change.removed;
-    let epoch_reward_budget = reward_pool
-        / Decimal::from_atomics(interval.epochs_in_interval(), 0).unwrap()
+    let epoch_reward_budget = reward_pool / interval.epochs_in_interval().into_base_decimal()?
         * rewarding_params.interval.interval_pool_emission;
     let stake_saturation_point =
-        staking_supply / Decimal::from_atomics(rewarding_params.rewarded_set_size, 0).unwrap();
+        staking_supply / rewarding_params.rewarded_set_size.into_base_decimal()?;
 
     rewarding_params.interval.reward_pool = reward_pool;
     rewarding_params.interval.staking_supply = staking_supply;
@@ -46,7 +46,7 @@ pub(crate) fn withdraw_operator_reward(
     let mix_id = mix_details.mix_id();
     let mut mix_rewarding = mix_details.rewarding_details;
     let original_pledge = mix_details.bond_information.original_pledge;
-    let reward = mix_rewarding.withdraw_operator_reward(&original_pledge);
+    let reward = mix_rewarding.withdraw_operator_reward(&original_pledge)?;
 
     // save updated rewarding info
     storage::MIXNODE_REWARDING.save(store, mix_id, &mix_rewarding)?;
@@ -87,7 +87,7 @@ mod tests {
         let mut test = TestSetup::new();
 
         let epochs_in_interval = test.current_interval().epochs_in_interval();
-        let epochs_in_interval_dec = Decimal::from_atomics(epochs_in_interval, 0).unwrap();
+        let epochs_in_interval_dec = epochs_in_interval.into_base_decimal().unwrap();
         let start_rewarding_params = test.rewarding_params();
 
         // nothing changes if pending changes are empty
@@ -95,7 +95,7 @@ mod tests {
         assert_eq!(start_rewarding_params, test.rewarding_params());
 
         // normal case of having distributed some rewards
-        let distributed_rewards = Decimal::from_atomics(100_000_000u32, 0).unwrap();
+        let distributed_rewards = 100_000_000u32.into_base_decimal().unwrap();
         storage::PENDING_REWARD_POOL_CHANGE
             .save(
                 test.deps_mut().storage,
@@ -132,7 +132,10 @@ mod tests {
         assert_eq!(
             updated_rewarding_params.interval.stake_saturation_point,
             updated_rewarding_params.interval.staking_supply
-                / Decimal::from_atomics(updated_rewarding_params.rewarded_set_size, 0).unwrap()
+                / updated_rewarding_params
+                    .rewarded_set_size
+                    .into_base_decimal()
+                    .unwrap()
         );
 
         // resets changes back to 0
@@ -144,7 +147,7 @@ mod tests {
         );
 
         // future case of having to also increase the reward pool
-        let added_credentials = Decimal::from_atomics(50_000_000u32, 0).unwrap();
+        let added_credentials = 50_000_000u32.into_base_decimal().unwrap();
         storage::PENDING_REWARD_POOL_CHANGE
             .save(
                 test.deps_mut().storage,
@@ -181,7 +184,10 @@ mod tests {
         assert_eq!(
             updated_rewarding_params2.interval.stake_saturation_point,
             updated_rewarding_params2.interval.staking_supply
-                / Decimal::from_atomics(updated_rewarding_params2.rewarded_set_size, 0).unwrap()
+                / updated_rewarding_params2
+                    .rewarded_set_size
+                    .into_base_decimal()
+                    .unwrap()
         );
 
         // resets changes back to 0
@@ -198,7 +204,7 @@ mod tests {
         let mut test = TestSetup::new();
 
         let pledge = Uint128::new(250_000_000);
-        let pledge_dec = Decimal::from_atomics(250_000_000u32, 0).unwrap();
+        let pledge_dec = 250_000_000u32.into_base_decimal().unwrap();
         let mix_id = test.add_dummy_mixnode("mix-owner", Some(pledge));
 
         // no rewards
@@ -235,7 +241,7 @@ mod tests {
         let mut test = TestSetup::new();
 
         let delegation_amount = Uint128::new(2500_000_000);
-        let delegation_dec = Decimal::from_atomics(2500_000_000u32, 0).unwrap();
+        let delegation_dec = 2500_000_000u32.into_base_decimal().unwrap();
         let mix_id = test.add_dummy_mixnode("mix-owner", None);
         let delegator = "delegator";
         test.add_immediate_delegation(delegator, delegation_amount, mix_id);

contracts/mixnet/src/rewards/queries.rs

@@ -7,6 +7,7 @@ use crate::interval::storage as interval_storage;
 use crate::mixnodes;
 use crate::mixnodes::storage as mixnodes_storage;
 use cosmwasm_std::{coin, Coin, Decimal, Deps, StdResult};
+use mixnet_contract_common::helpers::into_base_decimal;
 use mixnet_contract_common::mixnode::MixNodeDetails;
 use mixnet_contract_common::reward_params::{NodeRewardParams, Performance, RewardingParams};
 use mixnet_contract_common::rewarding::helpers::truncate_reward;
@@ -19,16 +20,18 @@ pub(crate) fn query_rewarding_params(deps: Deps<'_>) -> StdResult<RewardingParam
     storage::REWARDING_PARAMS.load(deps.storage)
 }
 
-fn pending_operator_reward(mix_details: Option<MixNodeDetails>) -> PendingRewardResponse {
-    match mix_details {
+fn pending_operator_reward(
+    mix_details: Option<MixNodeDetails>,
+) -> StdResult<PendingRewardResponse> {
+    Ok(match mix_details {
         Some(mix_details) => PendingRewardResponse {
             amount_staked: Some(mix_details.original_pledge().clone()),
             amount_earned: Some(mix_details.pending_operator_reward()),
-            amount_earned_detailed: Some(mix_details.pending_detailed_operator_reward()),
+            amount_earned_detailed: Some(mix_details.pending_detailed_operator_reward()?),
             mixnode_still_fully_bonded: !mix_details.is_unbonding(),
         },
         None => PendingRewardResponse::default(),
-    }
+    })
 }
 
 pub fn query_pending_operator_reward(
@@ -39,7 +42,7 @@ pub fn query_pending_operator_reward(
     // in order to determine operator's reward we need to know its original pledge and thus
     // we have to load the entire thing
     let mix_details = mixnodes::helpers::get_mixnode_details_by_owner(deps.storage, owner_address)?;
-    Ok(pending_operator_reward(mix_details))
+    pending_operator_reward(mix_details)
 }
 
 pub fn query_pending_mixnode_operator_reward(
@@ -49,7 +52,7 @@ pub fn query_pending_mixnode_operator_reward(
     // in order to determine operator's reward we need to know its original pledge and thus
     // we have to load the entire thing
     let mix_details = mixnodes::helpers::get_mixnode_details_by_id(deps.storage, mix_id)?;
-    Ok(pending_operator_reward(mix_details))
+    pending_operator_reward(mix_details)
 }
 
 pub fn query_pending_delegator_reward(
@@ -74,8 +77,8 @@ pub fn query_pending_delegator_reward(
         None => return Ok(PendingRewardResponse::default()),
     };
 
-    let detailed_reward = mix_rewarding.determine_delegation_reward(&delegation);
-    let delegator_reward = mix_rewarding.pending_delegator_reward(&delegation);
+    let detailed_reward = mix_rewarding.determine_delegation_reward(&delegation)?;
+    let delegator_reward = mix_rewarding.pending_delegator_reward(&delegation)?;
 
     // check if the mixnode isnt in the process of unbonding (or has already unbonded)
     let is_bonded = matches!(mixnodes_storage::mixnode_bonds().may_load(deps.storage, mix_id)?, Some(mix_bond) if !mix_bond.is_unbonding);
@@ -177,8 +180,8 @@ pub(crate) fn query_estimated_current_epoch_delegator_reward(
         None => return Ok(EstimatedCurrentEpochRewardResponse::empty_response()),
     };
 
-    let staked_dec = Decimal::from_atomics(delegation.amount.amount, 0).unwrap();
-    let current_value = staked_dec + mix_rewarding.determine_delegation_reward(&delegation);
+    let staked_dec = into_base_decimal(delegation.amount.amount)?;
+    let current_value = staked_dec + mix_rewarding.determine_delegation_reward(&delegation)?;
     let amount_staked = delegation.amount;
 
     // check if the mixnode isnt in the process of unbonding (or has already unbonded)
@@ -792,7 +795,10 @@ mod tests {
             let delegation = test.delegation(mix_id, owner, &None);
 
             let staked_dec = Decimal::from_atomics(delegation.amount.amount, 0).unwrap();
-            let current_value = staked_dec + mix_rewarding.determine_delegation_reward(&delegation);
+            let current_value = staked_dec
+                + mix_rewarding
+                    .determine_delegation_reward(&delegation)
+                    .unwrap();
             let amount_staked = delegation.amount;
 
             EstimatedCurrentEpochRewardResponse {

contracts/mixnet/src/rewards/transactions.rs

@@ -744,7 +744,7 @@ pub mod tests {
                         performance,
                         in_active_set: true,
                     };
-                    let sim_res = sim.simulate_epoch_single_node(node_params);
+                    let sim_res = sim.simulate_epoch_single_node(node_params).unwrap();
                     assert_eq!(sim_res, dist);
                 }
                 test.skip_to_next_epoch_end();
@@ -768,7 +768,7 @@ pub mod tests {
                         performance,
                         in_active_set: true,
                     };
-                    let sim_res = sim.simulate_epoch_single_node(node_params);
+                    let sim_res = sim.simulate_epoch_single_node(node_params).unwrap();
                     assert_eq!(sim_res, dist);
                 }
                 test.skip_to_next_epoch_end();
@@ -809,8 +809,8 @@ pub mod tests {
                     in_active_set: true,
                 };
 
-                let dist1 = sim1.simulate_epoch_single_node(node_params);
-                let dist2 = sim2.simulate_epoch_single_node(node_params);
+                let dist1 = sim1.simulate_epoch_single_node(node_params).unwrap();
+                let dist2 = sim2.simulate_epoch_single_node(node_params).unwrap();
 
                 let env = test.env();
 
@@ -858,15 +858,17 @@ pub mod tests {
                 let unit_delegation_base = actual_prior1.unit_delegation;
 
                 // recompute the state of fully compounded delegation from before this rewarding was distributed
-                let pre_rewarding_del11 = del11.dec_amount()
-                    + (prior_unit_reward - del11.cumulative_reward_ratio) * del11.dec_amount()
+                let pre_rewarding_del11 = del11.dec_amount().unwrap()
+                    + (prior_unit_reward - del11.cumulative_reward_ratio)
+                        * del11.dec_amount().unwrap()
                         / (del11.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del11_reward =
                     pre_rewarding_del11 / prior_delegates1 * delegates_reward1;
 
-                let pre_rewarding_del12 = del12.dec_amount()
-                    + (prior_unit_reward - del12.cumulative_reward_ratio) * del12.dec_amount()
+                let pre_rewarding_del12 = del12.dec_amount().unwrap()
+                    + (prior_unit_reward - del12.cumulative_reward_ratio)
+                        * del12.dec_amount().unwrap()
                         / (del12.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del12_reward =
@@ -920,8 +922,9 @@ pub mod tests {
                 let unit_delegation_base = actual_prior2.unit_delegation;
 
                 // recompute the state of fully compounded delegation from before this rewarding was distributed
-                let pre_rewarding_del21 = del21.dec_amount()
-                    + (prior_unit_reward - del21.cumulative_reward_ratio) * del21.dec_amount()
+                let pre_rewarding_del21 = del21.dec_amount().unwrap()
+                    + (prior_unit_reward - del21.cumulative_reward_ratio)
+                        * del21.dec_amount().unwrap()
                         / (del21.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del21_reward =
@@ -949,8 +952,8 @@ pub mod tests {
                     in_active_set: true,
                 };
 
-                let dist1 = sim1.simulate_epoch_single_node(node_params);
-                let dist2 = sim2.simulate_epoch_single_node(node_params);
+                let dist1 = sim1.simulate_epoch_single_node(node_params).unwrap();
+                let dist2 = sim2.simulate_epoch_single_node(node_params).unwrap();
 
                 let env = test.env();
 
@@ -998,22 +1001,25 @@ pub mod tests {
                 let unit_delegation_base = actual_prior1.unit_delegation;
 
                 // recompute the state of fully compounded delegation from before this rewarding was distributed
-                let pre_rewarding_del11 = del11.dec_amount()
-                    + (prior_unit_reward - del11.cumulative_reward_ratio) * del11.dec_amount()
+                let pre_rewarding_del11 = del11.dec_amount().unwrap()
+                    + (prior_unit_reward - del11.cumulative_reward_ratio)
+                        * del11.dec_amount().unwrap()
                         / (del11.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del11_reward =
                     pre_rewarding_del11 / prior_delegates1 * delegates_reward1;
 
-                let pre_rewarding_del12 = del12.dec_amount()
-                    + (prior_unit_reward - del12.cumulative_reward_ratio) * del12.dec_amount()
+                let pre_rewarding_del12 = del12.dec_amount().unwrap()
+                    + (prior_unit_reward - del12.cumulative_reward_ratio)
+                        * del12.dec_amount().unwrap()
                         / (del12.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del12_reward =
                     pre_rewarding_del12 / prior_delegates1 * delegates_reward1;
 
-                let pre_rewarding_del13 = del13.dec_amount()
-                    + (prior_unit_reward - del13.cumulative_reward_ratio) * del13.dec_amount()
+                let pre_rewarding_del13 = del13.dec_amount().unwrap()
+                    + (prior_unit_reward - del13.cumulative_reward_ratio)
+                        * del13.dec_amount().unwrap()
                         / (del13.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del13_reward =
@@ -1067,15 +1073,17 @@ pub mod tests {
                 let unit_delegation_base = actual_prior2.unit_delegation;
 
                 // recompute the state of fully compounded delegation from before this rewarding was distributed
-                let pre_rewarding_del21 = del21.dec_amount()
-                    + (prior_unit_reward - del21.cumulative_reward_ratio) * del21.dec_amount()
+                let pre_rewarding_del21 = del21.dec_amount().unwrap()
+                    + (prior_unit_reward - del21.cumulative_reward_ratio)
+                        * del21.dec_amount().unwrap()
                         / (del21.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del21_reward =
                     pre_rewarding_del21 / prior_delegates2 * delegates_reward2;
 
-                let pre_rewarding_del23 = del23.dec_amount()
-                    + (prior_unit_reward - del23.cumulative_reward_ratio) * del23.dec_amount()
+                let pre_rewarding_del23 = del23.dec_amount().unwrap()
+                    + (prior_unit_reward - del23.cumulative_reward_ratio)
+                        * del23.dec_amount().unwrap()
                         / (del23.cumulative_reward_ratio + unit_delegation_base);
 
                 let computed_del23_reward =

contracts/mixnet/src/support/helpers.rs

@@ -69,6 +69,10 @@ pub(crate) fn validate_pledge(
         });
     }
 
+    // throughout this function we've been using the value at `pledge[0]` without problems
+    // (plus we have even validated that the vec is not empty), so the unwrap here is absolutely fine,
+    // since it cannot possibly fail without UB
+    #[allow(clippy::unwrap_used)]
     Ok(pledge.pop().unwrap())
 }
 
@@ -106,6 +110,10 @@ pub(crate) fn validate_delegation_stake(
         return Err(MixnetContractError::EmptyDelegation);
     }
 
+    // throughout this function we've been using the value at `delegation[0]` without problems
+    // (plus we have even validated that the vec is not empty), so the unwrap here is absolutely fine,
+    // since it cannot possibly fail without UB
+    #[allow(clippy::unwrap_used)]
     Ok(delegation.pop().unwrap())
 }
 

contracts/vesting/src/contract.rs

@@ -1,8 +1,8 @@
 use crate::errors::ContractError;
 use crate::queued_migrations::migrate_to_v2_mixnet_contract;
 use crate::storage::{
-    account_from_address, BlockTimestampSecs, ADMIN, DELEGATIONS, MIXNET_CONTRACT_ADDRESS,
-    MIX_DENOM,
+    account_from_address, save_account, BlockTimestampSecs, ADMIN, DELEGATIONS,
+    MIXNET_CONTRACT_ADDRESS, MIX_DENOM,
 };
 use crate::traits::{
     DelegatingAccount, GatewayBondingAccount, MixnodeBondingAccount, VestingAccount,
@@ -158,7 +158,7 @@ pub fn try_update_locked_pledge_cap(
     let mut account = account_from_address(&address, deps.storage, deps.api)?;
 
     account.pledge_cap = Some(cap);
-    // update_locked_pledge_cap(amount, deps.storage)?;
+    save_account(&account, deps.storage)?;
     Ok(Response::default())
 }
 
@@ -589,7 +589,7 @@ pub fn try_get_current_vesting_period(
     env: Env,
 ) -> Result<Period, ContractError> {
     let account = account_from_address(address, deps.storage, deps.api)?;
-    Ok(account.get_current_vesting_period(env.block.time))
+    account.get_current_vesting_period(env.block.time)
 }
 
 /// Loads mixnode bond from vesting contract storage.
@@ -691,7 +691,7 @@ pub fn try_get_original_vesting(
     deps: Deps<'_>,
 ) -> Result<OriginalVestingResponse, ContractError> {
     let account = account_from_address(vesting_account_address, deps.storage, deps.api)?;
-    Ok(account.get_original_vesting())
+    account.get_original_vesting()
 }
 
 /// See [crate::traits::VestingAccount::get_delegated_free]

contracts/vesting/src/errors.rs

@@ -50,4 +50,6 @@ pub enum ContractError {
     MinVestingFunds { sent: u128, need: u128 },
     #[error("VESTING ({}): Maximum amount of locked coins has already been pledged: {current}, cap is {cap}", line!())]
     LockedPledgeCapReached { current: Uint128, cap: Uint128 },
+    #[error("VESTING: ({}: Account owned by {owner} has unpopulated vesting periods!", line!())]
+    UnpopulatedVestingPeriods { owner: Addr },
 }

contracts/vesting/src/lib.rs

@@ -1,6 +1,9 @@
 #![allow(rustdoc::private_intra_doc_links)]
 //! Nym vesting contract, providing vesting accounts with ability to stake unvested tokens
 
+#![warn(clippy::expect_used)]
+#![warn(clippy::unwrap_used)]
+
 pub mod contract;
 mod errors;
 mod queued_migrations;

contracts/vesting/src/traits/vesting_account.rs

@@ -55,7 +55,7 @@ pub trait VestingAccount {
 
     /// Returns amount of coins set at account creation
     /// See [/vesting-contract/struct.Account.html/method.get_original_vesting] for impl
-    fn get_original_vesting(&self) -> OriginalVestingResponse;
+    fn get_original_vesting(&self) -> Result<OriginalVestingResponse, ContractError>;
 
     /// See [/vesting-contract/struct.Account.html/method.get_delegated_free] for impl
     fn get_delegated_free(

contracts/vesting/src/vesting/account/mod.rs

@@ -69,7 +69,7 @@ impl Account {
     pub fn absolute_pledge_cap(&self) -> Result<Uint128, ContractError> {
         match self.pledge_cap() {
             PledgeCap::Absolute(cap) => Ok(cap),
-            PledgeCap::Percent(p) => Ok(p * self.get_original_vesting().amount.amount),
+            PledgeCap::Percent(p) => Ok(p * self.get_original_vesting()?.amount.amount),
         }
     }
 
@@ -81,8 +81,13 @@ impl Account {
         self.periods.len()
     }
 
-    pub fn period_duration(&self) -> u64 {
-        self.periods.get(0).unwrap().period_seconds
+    pub fn period_duration(&self) -> Result<u64, ContractError> {
+        self.periods
+            .get(0)
+            .ok_or(ContractError::UnpopulatedVestingPeriods {
+                owner: self.owner_address.clone(),
+            })
+            .map(|p| p.period_seconds)
     }
 
     pub fn storage_key(&self) -> u32 {
@@ -119,11 +124,28 @@ impl Account {
 
     /// Returns the index of the next vesting period. Unless the current time is somehow in the past or vesting has not started yet.
     /// In case vesting is over it will always return NUM_VESTING_PERIODS.
-    pub fn get_current_vesting_period(&self, block_time: Timestamp) -> Period {
-        if block_time.seconds() < self.periods.first().unwrap().start_time {
-            Period::Before
-        } else if self.periods.last().unwrap().end_time() < block_time {
-            Period::After
+    pub fn get_current_vesting_period(
+        &self,
+        block_time: Timestamp,
+    ) -> Result<Period, ContractError> {
+        let first_period =
+            self.periods
+                .first()
+                .ok_or(ContractError::UnpopulatedVestingPeriods {
+                    owner: self.owner_address.clone(),
+                })?;
+
+        let last_period = self
+            .periods
+            .last()
+            .ok_or(ContractError::UnpopulatedVestingPeriods {
+                owner: self.owner_address.clone(),
+            })?;
+
+        if block_time.seconds() < first_period.start_time {
+            Ok(Period::Before)
+        } else if last_period.end_time() < block_time {
+            Ok(Period::After)
         } else {
             let mut index = 0;
             for period in &self.periods {
@@ -132,7 +154,7 @@ impl Account {
                 }
                 index += 1;
             }
-            Period::In(index)
+            Ok(Period::In(index))
         }
     }