initial validator API-networking related things

adapted from DKG impl

.github/workflows/nightly_build.yml

@@ -75,6 +75,12 @@ jobs:
           command: clippy
           args: --workspace --all-targets -- -D warnings
 
+      - name: Reclaim some disk space (because Windows is being annoying)
+        uses: actions-rs/cargo@v1
+        if: ${{ matrix.os == 'windows-latest' }}
+        with:
+          command: clean
+
       # COCONUT stuff
       - name: Build all binaries with coconut enabled
         uses: actions-rs/cargo@v1

CHANGELOG.md

@@ -11,7 +11,7 @@
 - mixnet-contract: Replace all naked `-` with `saturating_sub`.
 - validator-api: add Swagger to document the REST API ([#1249]).
 - all: added network compilation target to `--help` (or `--version`) commands ([#1256]).
-- network-requester: send traffic statistics from all network requesters and receive it in a special network-requester that aggregates the data and exposes it via a rest API ([#1267]).
+- network-requester: send traffic statistics from all network requesters and receive it in a special network-requester that aggregates the data and exposes it via a rest API ([#1267], [#1278]).
 
 ### Fixed
 
@@ -28,6 +28,7 @@
 [#1265]: https://github.com/nymtech/nym/pull/1265
 [#1267]: https://github.com/nymtech/nym/pull/1267
 [#1275]: https://github.com/nymtech/nym/pull/1275
+[#1278]: https://github.com/nymtech/nym/pull/1278
 
 ## [nym-wallet-v1.0.4](https://github.com/nymtech/nym/tree/nym-wallet-v1.0.4) (2022-05-04)
 

Cargo.lock

@@ -1063,6 +1063,8 @@ dependencies = [
  "pemstore",
  "rand 0.7.3",
  "rand_chacha 0.2.2",
+ "serde",
+ "serde_bytes",
  "subtle-encoding",
  "x25519-dalek",
 ]
@@ -1427,6 +1429,7 @@ version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3d5c4b5e5959dc2c2b89918d8e2cc40fcdd623cef026ed09d2f0ee05199dc8e4"
 dependencies = [
+ "serde",
  "signature",
 ]
 
@@ -1440,6 +1443,7 @@ dependencies = [
  "ed25519",
  "rand 0.7.3",
  "serde",
+ "serde_bytes",
  "sha2",
  "zeroize",
 ]
@@ -4834,9 +4838,9 @@ dependencies = [
 
 [[package]]
 name = "serde_bytes"
-version = "0.11.5"
+version = "0.11.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "16ae07dd2f88a366f15bd0632ba725227018c69a1c8550a927324f8eb8368bb9"
+checksum = "212e73464ebcde48d723aa02eb270ba62eff38a9b732df31f33f1b4e145f3a54"
 dependencies = [
  "serde",
 ]

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

@@ -324,6 +324,7 @@ pub trait SigningCosmWasmClient: CosmWasmClient {
 
         Ok(ExecuteResult {
             logs: parse_raw_logs(tx_res.tx_result.log)?,
+            data: tx_res.tx_result.data,
             transaction_hash: tx_res.hash,
             gas_info,
         })
@@ -364,6 +365,7 @@ pub trait SigningCosmWasmClient: CosmWasmClient {
 
         Ok(ExecuteResult {
             logs: parse_raw_logs(tx_res.tx_result.log)?,
+            data: tx_res.tx_result.data,
             transaction_hash: tx_res.hash,
             gas_info,
         })

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

@@ -25,6 +25,7 @@ use cosmrs::proto::cosmwasm::wasm::v1::{
     CodeInfoResponse, ContractCodeHistoryEntry as ProtoContractCodeHistoryEntry,
     ContractCodeHistoryOperationType, ContractInfo as ProtoContractInfo,
 };
+use cosmrs::tendermint::abci::Data;
 use cosmrs::tendermint::{abci, chain};
 use cosmrs::tx::{AccountNumber, Gas, SequenceNumber};
 use cosmrs::{tx, AccountId, Any, Coin};
@@ -672,6 +673,8 @@ pub struct MigrateResult {
 pub struct ExecuteResult {
     pub logs: Vec<Log>,
 
+    pub data: Data,
+
     /// Transaction hash (might be used as transaction ID)
     pub transaction_hash: tx::Hash,
 

common/crypto/Cargo.toml

@@ -19,6 +19,8 @@ cipher = { version = "0.4.3", optional = true }
 x25519-dalek = { version = "1.1", optional = true }
 ed25519-dalek = { version = "1.0", optional = true }
 rand = { version = "0.7.3", features = ["wasm-bindgen"], optional = true }
+serde_bytes = { version = "0.11.6", optional = true }
+serde_crate = { version = "1.0", optional = true, default_features = false, package = "serde" }
 subtle-encoding = { version = "0.5", features =  ["bech32-preview"]}
 
 # internal
@@ -30,6 +32,7 @@ config = { path="../../common/config" }
 rand_chacha = "0.2"
 
 [features]
+serde = ["serde_crate", "serde_bytes", "ed25519-dalek/serde", "x25519-dalek/serde"]
 asymmetric = ["x25519-dalek", "ed25519-dalek"]
 hashing = ["blake3", "digest", "hkdf", "hmac", "generic-array"]
 symmetric = ["aes", "ctr", "cipher", "generic-array"]

common/crypto/src/asymmetric/encryption/mod.rs

@@ -4,6 +4,8 @@
 use pemstore::traits::{PemStorableKey, PemStorableKeyPair};
 #[cfg(feature = "rand")]
 use rand::{CryptoRng, RngCore};
+#[cfg(feature = "serde")]
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::fmt::{self, Display, Formatter};
 
 /// Size of a X25519 private key
@@ -127,6 +129,28 @@ impl PublicKey {
     }
 }
 
+#[cfg(feature = "serde")]
+impl Serialize for PublicKey {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+#[cfg(feature = "serde")]
+impl<'d> Deserialize<'d> for PublicKey {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'d>,
+    {
+        Ok(PublicKey(x25519_dalek::PublicKey::deserialize(
+            deserializer,
+        )?))
+    }
+}
+
 impl PemStorableKey for PublicKey {
     type Error = KeyRecoveryError;
 
@@ -143,7 +167,6 @@ impl PemStorableKey for PublicKey {
     }
 }
 
-#[derive(Clone)]
 pub struct PrivateKey(x25519_dalek::StaticSecret);
 
 impl Display for PrivateKey {
@@ -187,6 +210,28 @@ impl PrivateKey {
     }
 }
 
+#[cfg(feature = "serde")]
+impl Serialize for PrivateKey {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+#[cfg(feature = "serde")]
+impl<'d> Deserialize<'d> for PrivateKey {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'d>,
+    {
+        Ok(PrivateKey(x25519_dalek::StaticSecret::deserialize(
+            deserializer,
+        )?))
+    }
+}
+
 impl PemStorableKey for PrivateKey {
     type Error = KeyRecoveryError;
 

common/crypto/src/asymmetric/identity/mod.rs

@@ -10,6 +10,13 @@ use pemstore::traits::{PemStorableKey, PemStorableKeyPair};
 use rand::{CryptoRng, RngCore};
 use std::fmt::{self, Display, Formatter};
 
+#[cfg(feature = "serde")]
+use serde::de::Error as SerdeError;
+#[cfg(feature = "serde")]
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+#[cfg(feature = "serde")]
+use serde_bytes::{ByteBuf as SerdeByteBuf, Bytes as SerdeBytes};
+
 #[derive(Debug)]
 pub enum Ed25519RecoveryError {
     MalformedBytes(SignatureError),
@@ -40,6 +47,7 @@ impl fmt::Display for Ed25519RecoveryError {
 impl std::error::Error for Ed25519RecoveryError {}
 
 /// Keypair for usage in ed25519 EdDSA.
+#[derive(Debug)]
 pub struct KeyPair {
     private_key: PrivateKey,
     public_key: PublicKey,
@@ -135,6 +143,28 @@ impl PublicKey {
     }
 }
 
+#[cfg(feature = "serde")]
+impl Serialize for PublicKey {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+#[cfg(feature = "serde")]
+impl<'d> Deserialize<'d> for PublicKey {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'d>,
+    {
+        Ok(PublicKey(ed25519_dalek::PublicKey::deserialize(
+            deserializer,
+        )?))
+    }
+}
+
 impl PemStorableKey for PublicKey {
     type Error = Ed25519RecoveryError;
 
@@ -200,6 +230,28 @@ impl PrivateKey {
     }
 }
 
+#[cfg(feature = "serde")]
+impl Serialize for PrivateKey {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+#[cfg(feature = "serde")]
+impl<'d> Deserialize<'d> for PrivateKey {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'d>,
+    {
+        Ok(PrivateKey(ed25519_dalek::SecretKey::deserialize(
+            deserializer,
+        )?))
+    }
+}
+
 impl PemStorableKey for PrivateKey {
     type Error = Ed25519RecoveryError;
 
@@ -216,7 +268,7 @@ impl PemStorableKey for PrivateKey {
     }
 }
 
-#[derive(Debug)]
+#[derive(Copy, Clone, Debug)]
 pub struct Signature(ed25519_dalek::Signature);
 
 impl Signature {
@@ -237,3 +289,24 @@ impl Signature {
         Ok(Signature(ed25519_dalek::Signature::from_bytes(bytes)?))
     }
 }
+
+#[cfg(feature = "serde")]
+impl Serialize for Signature {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        SerdeBytes::new(&self.to_bytes()).serialize(serializer)
+    }
+}
+
+#[cfg(feature = "serde")]
+impl<'d> Deserialize<'d> for Signature {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'d>,
+    {
+        let bytes = <SerdeByteBuf>::deserialize(deserializer)?;
+        Signature::from_bytes(bytes.as_ref()).map_err(SerdeError::custom)
+    }
+}

common/crypto/src/lib.rs

@@ -29,5 +29,5 @@ pub use blake3;
 #[cfg(feature = "symmetric")]
 pub use ctr;
 
-// TODO: this function uses all three modules: asymmetric crypto, symmetric crypto and derives key...,
-// so I don't know where to put it...
+#[cfg(feature = "serde")]
+extern crate serde_crate as serde;

common/network-defaults/src/all.rs

@@ -56,6 +56,10 @@ impl Network {
         self.details().rewarding_validator_address
     }
 
+    pub fn stats_provider_network_address(&self) -> &str {
+        self.details().stats_provider_network_address
+    }
+
     pub fn validators(&self) -> impl Iterator<Item = &ValidatorDetails> {
         self.details().validators.iter()
     }

common/network-defaults/src/lib.rs

@@ -18,6 +18,7 @@ cfg_if::cfg_if! {
     if #[cfg(network = "mainnet")] {
         pub const DEFAULT_NETWORK: all::Network = all::Network::MAINNET;
         pub const DENOM: &str = mainnet::DENOM;
+        pub const STAKE_DENOM: &str = mainnet::STAKE_DENOM;
 
         pub const ETH_CONTRACT_ADDRESS: [u8; 20] = mainnet::_ETH_CONTRACT_ADDRESS;
         pub const ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] = mainnet::_ETH_ERC20_CONTRACT_ADDRESS;
@@ -25,6 +26,7 @@ cfg_if::cfg_if! {
     } else if #[cfg(network = "qa")] {
         pub const DEFAULT_NETWORK: all::Network = all::Network::QA;
         pub const DENOM: &str = qa::DENOM;
+        pub const STAKE_DENOM: &str = qa::STAKE_DENOM;
 
         pub const ETH_CONTRACT_ADDRESS: [u8; 20] = qa::_ETH_CONTRACT_ADDRESS;
         pub const ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] = qa::_ETH_ERC20_CONTRACT_ADDRESS;
@@ -32,6 +34,7 @@ cfg_if::cfg_if! {
     } else if #[cfg(network = "sandbox")] {
         pub const DEFAULT_NETWORK: all::Network = all::Network::SANDBOX;
         pub const DENOM: &str = sandbox::DENOM;
+        pub const STAKE_DENOM: &str = sandbox::STAKE_DENOM;
 
         pub const ETH_CONTRACT_ADDRESS: [u8; 20] = sandbox::_ETH_CONTRACT_ADDRESS;
         pub const ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] = sandbox::_ETH_ERC20_CONTRACT_ADDRESS;
@@ -49,6 +52,7 @@ pub struct DefaultNetworkDetails<'a> {
     vesting_contract_address: &'a str,
     bandwidth_claim_contract_address: &'a str,
     rewarding_validator_address: &'a str,
+    stats_provider_network_address: &'a str,
     validators: Vec<ValidatorDetails>,
 }
 
@@ -60,6 +64,7 @@ static MAINNET_DEFAULTS: Lazy<DefaultNetworkDetails<'static>> =
         vesting_contract_address: mainnet::VESTING_CONTRACT_ADDRESS,
         bandwidth_claim_contract_address: mainnet::BANDWIDTH_CLAIM_CONTRACT_ADDRESS,
         rewarding_validator_address: mainnet::REWARDING_VALIDATOR_ADDRESS,
+        stats_provider_network_address: mainnet::STATS_PROVIDER_CLIENT_ADDRESS,
         validators: mainnet::validators(),
     });
 
@@ -71,6 +76,7 @@ static SANDBOX_DEFAULTS: Lazy<DefaultNetworkDetails<'static>> =
         vesting_contract_address: sandbox::VESTING_CONTRACT_ADDRESS,
         bandwidth_claim_contract_address: sandbox::BANDWIDTH_CLAIM_CONTRACT_ADDRESS,
         rewarding_validator_address: sandbox::REWARDING_VALIDATOR_ADDRESS,
+        stats_provider_network_address: sandbox::STATS_PROVIDER_CLIENT_ADDRESS,
         validators: sandbox::validators(),
     });
 
@@ -81,6 +87,7 @@ static QA_DEFAULTS: Lazy<DefaultNetworkDetails<'static>> = Lazy::new(|| DefaultN
     vesting_contract_address: qa::VESTING_CONTRACT_ADDRESS,
     bandwidth_claim_contract_address: qa::BANDWIDTH_CLAIM_CONTRACT_ADDRESS,
     rewarding_validator_address: qa::REWARDING_VALIDATOR_ADDRESS,
+    stats_provider_network_address: qa::STATS_PROVIDER_CLIENT_ADDRESS,
     validators: qa::validators(),
 });
 

common/network-defaults/src/mainnet.rs

@@ -5,6 +5,7 @@ use crate::ValidatorDetails;
 
 pub(crate) const BECH32_PREFIX: &str = "n";
 pub const DENOM: &str = "unym";
+pub const STAKE_DENOM: &str = "unyx";
 
 pub(crate) const MIXNET_CONTRACT_ADDRESS: &str =
     "n14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9sjyvg3g";
@@ -18,6 +19,8 @@ pub(crate) const _ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] =
     hex_literal::hex!("0000000000000000000000000000000000000000");
 pub(crate) const REWARDING_VALIDATOR_ADDRESS: &str = "n10yyd98e2tuwu0f7ypz9dy3hhjw7v772q6287gy";
 
+pub(crate) const STATS_PROVIDER_CLIENT_ADDRESS: &str = "3V3me68qkEYNNShSQ5yLkrzC8rUJmcmtrTFbLKPqytEZ.7dGmnRAheEozNeGAsp9LXM8oPgS5YgJraNmYguj2t7Bn@BNjYZPxzcJwczXHHgBxCAyVJKxN6LPteDRrKapxWmexv";
+
 pub(crate) fn validators() -> Vec<ValidatorDetails> {
     vec![ValidatorDetails::new(
         "https://rpc.nyx.nodes.guru/",

common/network-defaults/src/qa.rs

@@ -5,6 +5,7 @@ use crate::ValidatorDetails;
 
 pub(crate) const BECH32_PREFIX: &str = "n";
 pub const DENOM: &str = "unym";
+pub const STAKE_DENOM: &str = "unyx";
 
 pub(crate) const MIXNET_CONTRACT_ADDRESS: &str =
     "n1suhgf5svhu4usrurvxzlgn54ksxmn8gljarjtxqnapv8kjnp4nrsd3qaep";
@@ -18,6 +19,8 @@ pub(crate) const _ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] =
     hex_literal::hex!("0000000000000000000000000000000000000000");
 pub(crate) const REWARDING_VALIDATOR_ADDRESS: &str = "n1tfzd4qz3a45u8p4mr5zmzv66457uwjgcl05jdq";
 
+pub(crate) const STATS_PROVIDER_CLIENT_ADDRESS: &str = "BLFPkyQ68xtR3TmrUWJZUKJF4SVwJR23wzQEmLHi2QcZ.5zms2X4ANsgY1VB4iC9kTqvbsHWmWUNSuvTtYr4Cp5qT@ExyJVqTSrgHTwzXm2r9RawfF5qYpvZjSVN2dLTs6bnWH";
+
 pub(crate) fn validators() -> Vec<ValidatorDetails> {
     vec![ValidatorDetails::new(
         "https://qa-validator.nymtech.net",

common/network-defaults/src/sandbox.rs

@@ -5,6 +5,7 @@ use crate::ValidatorDetails;
 
 pub(crate) const BECH32_PREFIX: &str = "nymt";
 pub const DENOM: &str = "unymt";
+pub const STAKE_DENOM: &str = "unyxt";
 
 pub(crate) const MIXNET_CONTRACT_ADDRESS: &str = "nymt1ghd753shjuwexxywmgs4xz7x2q732vcnstz02j";
 pub(crate) const VESTING_CONTRACT_ADDRESS: &str = "nymt14ejqjyq8um4p3xfqj74yld5waqljf88fn549lh";
@@ -16,6 +17,8 @@ pub(crate) const _ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] =
     hex_literal::hex!("E8883BAeF3869e14E4823F46662e81D4F7d2A81F");
 pub(crate) const REWARDING_VALIDATOR_ADDRESS: &str = "nymt1jh0s6qu6tuw9ut438836mmn7f3f2wencrnmdj4";
 
+pub(crate) const STATS_PROVIDER_CLIENT_ADDRESS: &str = "HqYWvCcB4sswYiyMj5Q8H5oc71kLf96vfrLK3npM7stH.CoeC5dcqurgdxr5zcgU77nZBSBCc8ntCiwUivQ9TX3KT@E3mvZTHQCdBvhfr178Swx9g4QG3kkRUun7YnToLMcMbM";
+
 pub(crate) fn validators() -> Vec<ValidatorDetails> {
     vec![ValidatorDetails::new(
         "https://sandbox-validator.nymtech.net",

explorer/src/components/MixNodes/index.ts

@@ -12,7 +12,7 @@ export type MixnodeRowType = {
   host: string;
   layer: string;
   profit_percentage: string;
-  avg_uptime: number;
+  avg_uptime: string;
 };
 
 export function mixnodeToGridRow(arrayOfMixnodes?: MixNodeResponse): MixnodeRowType[] {
@@ -36,6 +36,6 @@ export function mixNodeResponseItemToMixnodeRowType(item: MixNodeResponseItem):
     host: item?.mix_node?.host || '',
     layer: item?.layer || '',
     profit_percentage: `${profitPercentage}%`,
-    avg_uptime: item.avg_uptime,
+    avg_uptime: `${item.avg_uptime}%` || '-',
   };
 }

explorer/src/pages/Mixnodes/index.tsx

@@ -236,7 +236,7 @@ export const PageMixnodes: React.FC = () => {
       headerName: 'Average Uptime',
       renderHeader: () => <CustomColumnHeading headingTitle="Average Uptime" />,
       headerClassName: 'MuiDataGrid-header-override',
-      width: 140,
+      width: 160,
       headerAlign: 'left',
       renderCell: (params: GridRenderCellParams) => (
         <MuiLink

service-providers/network-requester/migrations/20220512120000_mixnet_statistics.sql

@@ -7,6 +7,7 @@ CREATE TABLE mixnet_statistics
 (
     id                         INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
     service_description        VARCHAR NOT NULL,
+    client_identity            VARCHAR NOT NULL,
     request_processed_bytes    INTEGER NOT NULL,
     response_processed_bytes   INTEGER NOT NULL,
     interval_seconds           INTEGER NOT NULL,

service-providers/network-requester/src/core.rs

@@ -10,10 +10,11 @@ use futures::channel::mpsc;
 use futures::stream::{SplitSink, SplitStream};
 use futures::{SinkExt, StreamExt};
 use log::*;
-use nymsphinx::addressing::clients::Recipient;
+use nymsphinx::addressing::clients::{ClientIdentity, Recipient};
 use nymsphinx::receiver::ReconstructedMessage;
 use proxy_helpers::connection_controller::{Controller, ControllerCommand, ControllerSender};
 use socks5_requests::{ConnectionId, Message as Socks5Message, Request, Response};
+use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
@@ -79,7 +80,7 @@ impl ServiceProvider {
             response_stats_data
                 .write()
                 .await
-                .processed(response.data.len() as u32);
+                .processed(return_address.identity(), response.data.len() as u32);
             // make 'request' to native-websocket client
             let response_message = ClientRequest::Send {
                 recipient: return_address,
@@ -226,6 +227,7 @@ impl ServiceProvider {
         controller_sender: &mut ControllerSender,
         mix_input_sender: &mpsc::UnboundedSender<(Response, Recipient)>,
         request_stats_data: &Arc<RwLock<StatsData>>,
+        connected_clients: &mut HashMap<ConnectionId, ClientIdentity>,
     ) {
         let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) {
             Ok(msg) => msg,
@@ -236,19 +238,24 @@ impl ServiceProvider {
         };
         match deserialized_msg {
             Socks5Message::Request(deserialized_request) => match deserialized_request {
-                Request::Connect(req) => self.handle_proxy_connect(
-                    controller_sender,
-                    mix_input_sender,
-                    req.conn_id,
-                    req.remote_addr,
-                    req.return_address,
-                ),
+                Request::Connect(req) => {
+                    connected_clients.insert(req.conn_id, *req.return_address.identity());
+                    self.handle_proxy_connect(
+                        controller_sender,
+                        mix_input_sender,
+                        req.conn_id,
+                        req.remote_addr,
+                        req.return_address,
+                    )
+                }
 
                 Request::Send(conn_id, data, closed) => {
-                    request_stats_data
-                        .write()
-                        .await
-                        .processed(data.len() as u32);
+                    if let Some(client_identity) = connected_clients.get(&conn_id) {
+                        request_stats_data
+                            .write()
+                            .await
+                            .processed(client_identity, data.len() as u32);
+                    }
                     self.handle_proxy_send(controller_sender, conn_id, data, closed)
                 }
             },
@@ -329,6 +336,7 @@ impl ServiceProvider {
 
         println!("\nAll systems go. Press CTRL-C to stop the server.");
         // for each incoming message from the websocket... (which in 99.99% cases is going to be a mix message)
+        let mut connected_clients = HashMap::new();
         loop {
             let received = match Self::read_websocket_message(&mut websocket_reader).await {
                 Some(msg) => msg,
@@ -348,6 +356,7 @@ impl ServiceProvider {
                 &mut controller_sender,
                 &mix_input_sender,
                 &request_stats_data,
+                &mut connected_clients,
             )
             .await;
         }

service-providers/network-requester/src/statistics/comm.rs

@@ -6,25 +6,21 @@ use futures::StreamExt;
 use log::*;
 use serde::{Deserialize, Serialize};
 use sqlx::types::chrono::{DateTime, Utc};
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
 
-use nymsphinx::addressing::clients::{ClientEncryptionKey, ClientIdentity, Recipient};
-use nymsphinx::addressing::nodes::NodeIdentity;
+use network_defaults::DEFAULT_NETWORK;
+use nymsphinx::addressing::clients::{ClientIdentity, Recipient};
 use socks5_requests::Response;
 
 use super::error::StatsError;
 
-const STATS_PROVIDER_CLIENT_IDENTITY: &str = "HqYWvCcB4sswYiyMj5Q8H5oc71kLf96vfrLK3npM7stH";
-const STATS_PROVIDER_ENCRYPTION_KEY: &str = "CoeC5dcqurgdxr5zcgU77nZBSBCc8ntCiwUivQ9TX3KT";
-const STATS_PROVIDER_GATEWAY_IDENTITY: &str = "E3mvZTHQCdBvhfr178Swx9g4QG3kkRUun7YnToLMcMbM";
-
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct StatsMessage {
     pub description: String,
-    pub request_data: StatsData,
-    pub response_data: StatsData,
+    pub stats_data: Vec<StatsClientData>,
     pub interval_seconds: u32,
     pub timestamp: String,
 }
@@ -41,23 +37,42 @@ impl StatsMessage {
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct StatsClientData {
+    pub client_identity: String,
+    pub request_bytes: u32,
+    pub response_bytes: u32,
+}
+
+impl StatsClientData {
+    pub fn new(client_identity: String, request_bytes: u32, response_bytes: u32) -> Self {
+        StatsClientData {
+            client_identity,
+            request_bytes,
+            response_bytes,
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
 pub struct StatsData {
-    total_processed_bytes: u32,
+    client_processed_bytes: HashMap<String, u32>,
 }
 
 impl StatsData {
-    pub fn new(total_processed_bytes: u32) -> Self {
+    pub fn new() -> Self {
         StatsData {
-            total_processed_bytes,
+            client_processed_bytes: HashMap::new(),
         }
     }
-    pub fn processed(&mut self, bytes: u32) {
-        self.total_processed_bytes += bytes;
-    }
 
-    #[cfg(feature = "stats-service")]
-    pub fn total_processed_bytes(&self) -> u32 {
-        self.total_processed_bytes
+    pub fn processed(&mut self, client_identity: &ClientIdentity, bytes: u32) {
+        let client_identity_bs58 = client_identity.to_base58_string();
+        if let Some(curr_bytes) = self.client_processed_bytes.get_mut(&client_identity_bs58) {
+            *curr_bytes += bytes;
+        } else {
+            self.client_processed_bytes
+                .insert(client_identity_bs58, bytes);
+        }
     }
 }
 
@@ -77,16 +92,14 @@ impl Statistics {
         interval_seconds: Duration,
         timer_receiver: mpsc::Receiver<()>,
     ) -> Self {
-        // those unwraps are ok because we set the strings in the constants above
-        let stats_provider_addr = Recipient::new(
-            ClientIdentity::from_base58_string(STATS_PROVIDER_CLIENT_IDENTITY).unwrap(),
-            ClientEncryptionKey::from_base58_string(STATS_PROVIDER_ENCRYPTION_KEY).unwrap(),
-            NodeIdentity::from_base58_string(STATS_PROVIDER_GATEWAY_IDENTITY).unwrap(),
-        );
+        // this unwrap is ok because we set the string in a constant
+        let stats_provider_addr =
+            Recipient::try_from_base58_string(DEFAULT_NETWORK.stats_provider_network_address())
+                .unwrap();
         Statistics {
             description,
-            request_data: Arc::new(RwLock::new(StatsData::new(0))),
-            response_data: Arc::new(RwLock::new(StatsData::new(0))),
+            request_data: Arc::new(RwLock::new(StatsData::new())),
+            response_data: Arc::new(RwLock::new(StatsData::new())),
             timestamp: Utc::now(),
             interval_seconds: interval_seconds.as_secs() as u32,
             timer_receiver,
@@ -107,10 +120,36 @@ impl Statistics {
             if self.timer_receiver.next().await == None {
                 error!("Timer thread has died. No more statistics will be sent");
             } else {
+                let stats_data = {
+                    let request_data_bytes = self.request_data.read().await;
+                    let response_data_bytes = self.response_data.read().await;
+                    let clients: HashSet<String> = request_data_bytes
+                        .client_processed_bytes
+                        .keys()
+                        .chain(response_data_bytes.client_processed_bytes.keys())
+                        .cloned()
+                        .collect();
+                    clients
+                        .into_iter()
+                        .map(|client_identity| {
+                            let request_bytes = request_data_bytes
+                                .client_processed_bytes
+                                .get(&client_identity)
+                                .copied()
+                                .unwrap_or(0);
+                            let response_bytes = response_data_bytes
+                                .client_processed_bytes
+                                .get(&client_identity)
+                                .copied()
+                                .unwrap_or(0);
+                            StatsClientData::new(client_identity, request_bytes, response_bytes)
+                        })
+                        .collect()
+                };
+
                 let stats_message = StatsMessage {
                     description: self.description.clone(),
-                    request_data: self.request_data.read().await.clone(),
-                    response_data: self.response_data.read().await.clone(),
+                    stats_data,
                     interval_seconds: self.interval_seconds,
                     timestamp: self.timestamp.to_rfc3339(),
                 };
@@ -132,8 +171,18 @@ impl Statistics {
     }
 
     async fn reset_stats(&mut self) {
-        self.request_data.write().await.total_processed_bytes = 0;
-        self.response_data.write().await.total_processed_bytes = 0;
+        self.request_data
+            .write()
+            .await
+            .client_processed_bytes
+            .iter_mut()
+            .for_each(|(_, b)| *b = 0);
+        self.response_data
+            .write()
+            .await
+            .client_processed_bytes
+            .iter_mut()
+            .for_each(|(_, b)| *b = 0);
         self.timestamp = Utc::now();
     }
 }

service-providers/network-requester/src/statistics/mod.rs

@@ -5,5 +5,5 @@ mod comm;
 mod error;
 mod timer;
 
-pub use comm::{Statistics, StatsData, StatsMessage};
+pub use comm::{Statistics, StatsClientData, StatsData, StatsMessage};
 pub use timer::Timer;

service-providers/network-requester/src/storage/manager.rs

@@ -17,25 +17,31 @@ impl StorageManager {
     /// # Arguments
     ///
     /// * `service_description`: Description of the service that gathered the data.
+    /// * `client_identity`: Client that connected to the service.
     /// * `request_processed_bytes`: Number of bytes for socks5 requests.
     /// * `response_processed_bytes`: Number of bytes for socks5 responses.
     /// * `interval_seconds`: Duration in seconds in which the data was gathered.
     pub(super) async fn insert_service_statistics(
         &self,
         service_description: String,
+        client_identity: String,
         request_processed_bytes: u32,
         response_processed_bytes: u32,
         interval_seconds: u32,
         timestamp: DateTime<Utc>,
     ) -> Result<(), sqlx::Error> {
         sqlx::query!(
-            "INSERT INTO mixnet_statistics(service_description, request_processed_bytes, response_processed_bytes, interval_seconds, timestamp) VALUES (?, ?, ?, ?, ?)",
-                service_description,
-                request_processed_bytes,
-                response_processed_bytes,
-                interval_seconds,
-                timestamp,
-            ).execute(&self.connection_pool).await?;
+            "INSERT INTO mixnet_statistics(service_description, client_identity, request_processed_bytes, response_processed_bytes, interval_seconds, timestamp) VALUES (?, ?, ?, ?, ?, ?)",
+            service_description,
+            client_identity,
+            request_processed_bytes,
+            response_processed_bytes,
+            interval_seconds,
+            timestamp,
+        )
+        .execute(&self.connection_pool)
+        .await?;
+
         Ok(())
     }
 

service-providers/network-requester/src/storage/mod.rs

@@ -55,16 +55,20 @@ impl NetworkRequesterStorage {
         let timestamp: DateTime<Utc> = DateTime::parse_from_rfc3339(&msg.timestamp)
             .map_err(|_| NetworkRequesterStorageError::TimestampParse)?
             .into();
-        Ok(self
-            .manager
-            .insert_service_statistics(
-                msg.description,
-                msg.request_data.total_processed_bytes(),
-                msg.response_data.total_processed_bytes(),
-                msg.interval_seconds,
-                timestamp,
-            )
-            .await?)
+        for client_data in msg.stats_data {
+            self.manager
+                .insert_service_statistics(
+                    msg.description.clone(),
+                    client_data.client_identity.clone(),
+                    client_data.request_bytes,
+                    client_data.response_bytes,
+                    msg.interval_seconds,
+                    timestamp,
+                )
+                .await?;
+        }
+
+        Ok(())
     }
 
     /// Returns data submitted within the provided time interval.

service-providers/network-requester/src/storage/models.rs

@@ -8,6 +8,7 @@ pub(crate) struct MixnetStatistics {
     #[allow(dead_code)]
     pub(crate) id: i64,
     pub(crate) service_description: String,
+    pub(crate) client_identity: String,
     pub(crate) request_processed_bytes: i64,
     pub(crate) response_processed_bytes: i64,
     pub(crate) interval_seconds: i64,

service-providers/network-requester/src/storage/routes.rs

@@ -5,7 +5,6 @@ use rocket::serde::json::Json;
 use rocket::State;
 use serde::{Deserialize, Serialize};
 
-use crate::statistics::{StatsData, StatsMessage};
 use crate::storage::NetworkRequesterStorage;
 
 #[derive(Clone, Serialize, Deserialize, Debug)]
@@ -18,14 +17,19 @@ pub struct MixnetStatisticsRequest {
 
 #[derive(Clone, Serialize, Deserialize, Debug)]
 pub struct MixnetStatisticsResponse {
-    data: Vec<StatsMessage>,
+    pub service_description: String,
+    pub client_identity: String,
+    pub request_processed_bytes: u32,
+    pub response_processed_bytes: u32,
+    pub interval_seconds: u32,
+    pub timestamp: String,
 }
 
 #[rocket::post("/mixnet-statistics", data = "<mixnet_statistics_request>")]
 pub(crate) async fn post_mixnet_statistics(
     mixnet_statistics_request: Json<MixnetStatisticsRequest>,
     storage: &State<NetworkRequesterStorage>,
-) -> Json<MixnetStatisticsResponse> {
+) -> Json<Vec<MixnetStatisticsResponse>> {
     let mixnet_statistics = storage
         .get_service_statistics_in_interval(
             &mixnet_statistics_request.since,
@@ -34,16 +38,15 @@ pub(crate) async fn post_mixnet_statistics(
         .await
         .unwrap()
         .into_iter()
-        .map(|data| StatsMessage {
-            description: data.service_description,
-            request_data: StatsData::new(data.request_processed_bytes as u32),
-            response_data: StatsData::new(data.response_processed_bytes as u32),
+        .map(|data| MixnetStatisticsResponse {
+            service_description: data.service_description,
+            client_identity: data.client_identity,
+            request_processed_bytes: data.request_processed_bytes as u32,
+            response_processed_bytes: data.response_processed_bytes as u32,
             interval_seconds: data.interval_seconds as u32,
             timestamp: data.timestamp.to_string(),
         })
         .collect();
 
-    Json(MixnetStatisticsResponse {
-        data: mixnet_statistics,
-    })
+    Json(mixnet_statistics)
 }

validator-api/src/main.rs

@@ -45,6 +45,9 @@ mod rewarded_set_updater;
 pub(crate) mod storage;
 mod swagger;
 
+#[allow(dead_code)]
+mod networking;
+
 #[cfg(feature = "coconut")]
 mod coconut;
 
@@ -590,6 +593,8 @@ async fn run_validator_api(matches: ArgMatches<'static>) -> Result<()> {
 async fn main() -> Result<()> {
     println!("Starting validator api...");
 
+    dotenv::dotenv()?;
+
     cfg_if::cfg_if! {if #[cfg(feature = "console-subscriber")] {
         // instriment tokio console subscriber needs RUSTFLAGS="--cfg tokio_unstable" at build time
         console_subscriber::init();

validator-api/src/networking/codec.rs

@@ -0,0 +1,65 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::error::NetworkingError;
+use crate::networking::message::{Header, OffchainMessage};
+use crate::networking::PROTOCOL_VERSION;
+use bytes::{Buf, BytesMut};
+use tokio_util::codec::{Decoder, Encoder};
+
+// TODO: that was fine for the purposes of DKG, we might want to increase it if it's used anywhere else
+const MAX_ALLOWED_MESSAGE_LEN: u64 = 2 * 1024 * 1024;
+
+#[derive(Debug)]
+pub struct OffchainCodec;
+
+impl<'a> Encoder<&'a OffchainMessage> for OffchainCodec {
+    type Error = NetworkingError;
+
+    fn encode(&mut self, item: &OffchainMessage, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        item.encode(dst)
+    }
+}
+
+impl Decoder for OffchainCodec {
+    type Item = OffchainMessage;
+    type Error = NetworkingError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.len() < Header::LEN {
+            // can't do much without being able to deserialize the header
+            return Ok(None);
+        }
+
+        // header deserialization is simple enough to not be too cumbersome if it were to be repeated
+        let header = Header::try_from_bytes(&src[..Header::LEN])?;
+
+        if header.payload_length > MAX_ALLOWED_MESSAGE_LEN {
+            return Err(NetworkingError::MessageTooLarge {
+                supported: MAX_ALLOWED_MESSAGE_LEN,
+                received: header.payload_length,
+            });
+        }
+
+        if header.protocol_version != PROTOCOL_VERSION {
+            return Err(NetworkingError::MismatchedProtocolVersion {
+                expected: PROTOCOL_VERSION,
+                received: header.protocol_version,
+            });
+        }
+
+        if src.len() < Header::LEN + header.payload_length as usize {
+            // we haven't received the entire expected message yet.
+            // However, reserve enough bytes in the buffer for it
+            src.reserve(Header::LEN + header.payload_length as usize - src.len());
+
+            // We inform the Framed that we need more bytes to form the next frame.
+            return Ok(None);
+        }
+
+        let payload = src[Header::LEN..Header::LEN + header.payload_length as usize].to_vec();
+        src.advance(Header::LEN + header.payload_length as usize);
+
+        Ok(Some(OffchainMessage::try_from_bytes(payload)?))
+    }
+}

validator-api/src/networking/error.rs

@@ -0,0 +1,22 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::io;
+use thiserror::Error;
+
+// this would probably need to adjusted or maybe incorporated into existing stuff as originally
+// this error was more extensive by being a global `DkgError`
+#[derive(Error, Debug)]
+pub enum NetworkingError {
+    #[error("Networking / IO error - {0}")]
+    Io(#[from] io::Error),
+
+    #[error("Received message with specified size bigger than the supported maximum.  Received: {received}, supported: {supported}")]
+    MessageTooLarge { supported: u64, received: u64 },
+
+    #[error("Received message with unexpected protocol version. Received: {received}, expected: {expected}")]
+    MismatchedProtocolVersion { expected: u32, received: u32 },
+
+    #[error("Failed to deal with serialization (or deserialization) of the message - {0}")]
+    SerializationError(#[from] bincode::Error),
+}

validator-api/src/networking/message.rs

@@ -0,0 +1,165 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::error::NetworkingError;
+use crate::networking::PROTOCOL_VERSION;
+use bytes::{BufMut, BytesMut};
+use serde::{Deserialize, Serialize};
+use std::fmt::{Display, Formatter};
+use std::io;
+use std::time::Duration;
+use thiserror::Error;
+
+// I left a sample `NewDealingMessage` to show how it was originally implemented
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum OffchainMessage {
+    // you'd add something like this:
+    // NewDealing {
+    //     id: u64,
+    //     message: NewDealingMessage,
+    // },
+    ErrorResponse {
+        id: Option<u64>,
+        message: ErrorResponseMessage,
+    },
+}
+
+impl Display for OffchainMessage {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "OffchainMessage ")?;
+        match self {
+            // OffchainDkgMessage::NewDealing { id, message } => {
+            //     write!(f, "with id {} and message: {}", id, message)
+            // }
+            OffchainMessage::ErrorResponse { id, message } => {
+                write!(f, "with id {:?} and message: {}", id, message)
+            }
+        }
+    }
+}
+
+// #[derive(Debug, Clone, Serialize, Deserialize)]
+// pub struct NewDealingMessage {
+//     pub epoch_id: u32,
+//     pub dealing_bytes: Vec<u8>,
+//     pub dealer_signature: identity::Signature,
+// }
+// impl Display for NewDealingMessage {
+//     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+//         write!(
+//             f,
+//             "NewDealingMessage for epoch {} with length {}",
+//             self.epoch_id,
+//             self.dealing_bytes.len()
+//         )
+//     }
+// }
+
+#[derive(Debug, Clone, Serialize, Deserialize, Error)]
+pub enum ErrorResponseMessage {
+    #[error("{typ} is not a valid request type")]
+    InvalidRequest { typ: String },
+
+    #[error("This request failed to get resolved within {} seconds", .timeout.as_secs())]
+    Timeout { timeout: Duration },
+}
+
+impl OffchainMessage {
+    pub(crate) fn new_error_response(id: Option<u64>, message: ErrorResponseMessage) -> Self {
+        OffchainMessage::ErrorResponse { id, message }
+    }
+
+    pub(crate) fn try_from_bytes(bytes: Vec<u8>) -> Result<Self, NetworkingError> {
+        Ok(bincode::deserialize(&bytes)?)
+    }
+
+    pub(crate) fn try_to_bytes(&self) -> Result<Vec<u8>, NetworkingError> {
+        Ok(bincode::serialize(&self)?)
+    }
+
+    fn frame(&self) -> Result<FramedOffchainDkgMessage, NetworkingError> {
+        let payload = self.try_to_bytes()?;
+        Ok(FramedOffchainDkgMessage {
+            header: Header {
+                payload_length: payload.len() as u64,
+                protocol_version: PROTOCOL_VERSION,
+            },
+            payload,
+        })
+    }
+
+    pub(crate) fn encode(&self, dst: &mut BytesMut) -> Result<(), NetworkingError> {
+        dst.put(self.frame()?.into_bytes().as_ref());
+        Ok(())
+    }
+}
+
+struct FramedOffchainDkgMessage {
+    header: Header,
+    payload: Vec<u8>,
+}
+
+impl FramedOffchainDkgMessage {
+    fn into_bytes(mut self) -> Vec<u8> {
+        let mut header_bytes = self.header.into_bytes();
+        let mut out = Vec::with_capacity(header_bytes.len() + self.payload.len());
+
+        out.append(&mut header_bytes);
+        out.append(&mut self.payload);
+        out
+    }
+}
+
+#[derive(Debug, Copy, Clone, PartialEq)]
+pub(crate) struct Header {
+    pub(crate) payload_length: u64,
+    pub(crate) protocol_version: u32,
+}
+
+impl Header {
+    pub(crate) const LEN: usize = 12;
+
+    pub(crate) fn into_bytes(self) -> Vec<u8> {
+        let mut out = Vec::with_capacity(Self::LEN);
+        out.extend_from_slice(&self.payload_length.to_be_bytes());
+        out.extend_from_slice(&self.protocol_version.to_be_bytes());
+
+        debug_assert_eq!(Self::LEN, out.len());
+        out
+    }
+
+    pub(crate) fn try_from_bytes(bytes: &[u8]) -> Result<Self, NetworkingError> {
+        if bytes.len() != Self::LEN {
+            return Err(NetworkingError::Io(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "OffchainMessageType::Header: got {} bytes, expected: {}",
+                    bytes.len(),
+                    Self::LEN
+                ),
+            )));
+        }
+        Ok(Header {
+            payload_length: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
+            protocol_version: u32::from_be_bytes(bytes[8..].try_into().unwrap()),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::networking::PROTOCOL_VERSION;
+
+    #[test]
+    fn header_deserialization() {
+        let valid_header = Header {
+            payload_length: 1234,
+            protocol_version: PROTOCOL_VERSION,
+        };
+
+        let bytes = valid_header.into_bytes();
+        assert_eq!(valid_header, Header::try_from_bytes(&bytes).unwrap())
+    }
+}

validator-api/src/networking/mod.rs

@@ -0,0 +1,14 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+// TODO: if this becomes too cumbersome, perhaps consider a more streamlined solution like tarpc
+// (I wouldn't have needed that for DKG, but if this is to be used for different purposes, maybe
+// it would have been more appropriate)
+
+pub(crate) mod codec;
+pub(crate) mod error;
+pub(crate) mod message;
+pub(crate) mod receiver;
+pub(crate) mod sender;
+
+pub(crate) const PROTOCOL_VERSION: u32 = 1;

validator-api/src/networking/receiver/handler.rs

@@ -0,0 +1,98 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::codec::OffchainCodec;
+use crate::networking::message::{ErrorResponseMessage, OffchainMessage};
+// use crate::dkg::state::StateAccessor;
+use futures::{SinkExt, StreamExt};
+use std::net::SocketAddr;
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::time::timeout;
+use tokio_util::codec::Framed;
+
+const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::from_secs(2 * 60 * 60);
+
+#[derive(Debug)]
+pub struct ConnectionHandler {
+    // connection cannot exist for more than this time
+    max_connection_duration: Duration,
+    // state_accessor: StateAccessor,
+    conn: Framed<TcpStream, OffchainCodec>,
+    remote: SocketAddr,
+}
+
+impl ConnectionHandler {
+    pub(crate) fn new(conn: TcpStream, remote: SocketAddr) -> Self {
+        ConnectionHandler {
+            max_connection_duration: DEFAULT_MAX_CONNECTION_DURATION,
+            remote,
+            conn: Framed::new(conn, OffchainCodec),
+        }
+    }
+
+    async fn send_response(&mut self, response_message: OffchainMessage) {
+        if let Err(err) = self.conn.send(&response_message).await {
+            warn!("Failed to send response back to {} - {}", self.remote, err)
+        }
+    }
+
+    async fn send_error_response(&mut self, id: Option<u64>, error: ErrorResponseMessage) {
+        self.send_response(OffchainMessage::new_error_response(id, error))
+            .await
+    }
+
+    async fn handle_request(&mut self, request: OffchainMessage) {
+        match request {
+            OffchainMessage::ErrorResponse { id, .. } => {
+                self.send_error_response(
+                    id,
+                    ErrorResponseMessage::InvalidRequest {
+                        typ: "ErrorResponse".into(),
+                    },
+                )
+                .await
+            }
+        }
+    }
+
+    async fn _handle_connection(&mut self) {
+        debug!("Starting connection handler for {}", self.remote);
+
+        while let Some(framed_dkg_request) = self.conn.next().await {
+            trace!("received new message from {}", self.remote);
+            match framed_dkg_request {
+                Ok(framed_dkg_request) => self.handle_request(framed_dkg_request).await,
+                Err(err) => {
+                    warn!(
+                        "The socket connection got corrupted with error: {:?}. Closing the socket",
+                        err
+                    );
+                    break;
+                }
+            }
+        }
+
+        debug!("Closing connection from {}", self.remote);
+    }
+
+    pub async fn handle_connection(mut self) {
+        let remote = self.remote;
+        if timeout(self.max_connection_duration, self._handle_connection())
+            .await
+            .is_err()
+        {
+            warn!(
+                "we timed out while trying to resolve connection from {}",
+                remote
+            );
+            self.send_error_response(
+                None,
+                ErrorResponseMessage::Timeout {
+                    timeout: self.max_connection_duration,
+                },
+            )
+            .await;
+        }
+    }
+}

validator-api/src/networking/receiver/listener.rs

@@ -0,0 +1,47 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::receiver::handler::ConnectionHandler;
+use log::debug;
+use std::fmt::Display;
+use std::net::SocketAddr;
+use std::process;
+use tokio::net::{TcpListener, TcpStream, ToSocketAddrs};
+
+// note that we do not expect persistent connections between dealers, they should only really
+// exist for the duration of a single message exchange
+pub struct Listener<A> {
+    address: A,
+}
+
+impl<A> Listener<A> {
+    pub(crate) fn new(address: A) -> Self {
+        Listener { address }
+    }
+
+    fn on_connect(&self, conn: TcpStream, remote: SocketAddr) {
+        tokio::spawn(ConnectionHandler::new(conn, remote).handle_connection());
+    }
+
+    pub(crate) async fn run(&mut self)
+    where
+        A: ToSocketAddrs + Display,
+    {
+        debug!("starting off-chain DKG Listener");
+
+        let listener = match TcpListener::bind(&self.address).await {
+            Ok(listener) => listener,
+            Err(err) => {
+                error!("Failed to bind to {} - {}.", self.address, err);
+                process::exit(1);
+            }
+        };
+
+        loop {
+            match listener.accept().await {
+                Ok((socket, remote_addr)) => self.on_connect(socket, remote_addr),
+                Err(err) => warn!("Failed to accept incoming connection - {:?}", err),
+            }
+        }
+    }
+}

validator-api/src/networking/receiver/mod.rs

@@ -0,0 +1,7 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+mod handler;
+mod listener;
+
+pub(crate) use listener::Listener;

validator-api/src/networking/sender/broadcast.rs

@@ -0,0 +1,122 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::message::OffchainMessage;
+use crate::networking::sender::{send_single_message, ConnectionConfig, SendResponse};
+use futures::channel::mpsc;
+use futures::{stream, StreamExt};
+use std::net::SocketAddr;
+
+// TODO: for now just leave it here and make it configurable with proper config later
+const DEFAULT_CONCURRENCY: usize = 5;
+
+type FeedbackSender = mpsc::UnboundedSender<SendResponse>;
+
+pub(crate) struct Broadcaster {
+    addresses: Vec<SocketAddr>,
+    concurrency_level: usize,
+    config: ConnectionConfig,
+}
+
+impl Broadcaster {
+    pub(crate) fn new(addresses: Vec<SocketAddr>, config: ConnectionConfig) -> Self {
+        Broadcaster {
+            addresses,
+            concurrency_level: DEFAULT_CONCURRENCY,
+            config,
+        }
+    }
+
+    pub(crate) fn with_concurrency_level(mut self, concurrency_level: usize) -> Self {
+        self.concurrency_level = concurrency_level;
+        self
+    }
+
+    pub(crate) fn set_addresses(&mut self, new_addresses: Vec<SocketAddr>) {
+        self.addresses = new_addresses;
+    }
+
+    fn create_broadcast_configs(
+        &self,
+        message: OffchainMessage,
+        feedback_sender: Option<FeedbackSender>,
+    ) -> Vec<BroadcastConfig> {
+        self.addresses
+            .iter()
+            .map(|&address| BroadcastConfig {
+                address,
+                config: self.config,
+                feedback_sender: feedback_sender.clone(),
+                message: message.clone(),
+            })
+            .collect()
+    }
+
+    pub(crate) async fn broadcast_with_feedback(&self, msg: OffchainMessage) -> Vec<SendResponse> {
+        if self.addresses.is_empty() {
+            warn!("attempting to broadcast {} while no remotes are known", msg);
+            return Vec::new();
+        }
+
+        debug!("broadcasting {} to {} remotes", msg, self.addresses.len());
+        let (feedback_tx, mut feedback_rx) = mpsc::unbounded();
+
+        stream::iter(self.create_broadcast_configs(msg, Some(feedback_tx)))
+            .for_each_concurrent(self.concurrency_level, |cfg| cfg.send())
+            .await;
+
+        let mut responses = Vec::new();
+
+        for _ in 0..self.addresses.len() {
+            // we should have received exactly self.addresses number of responses
+            // (they could be just Err failure responses, but should exist nonetheless)
+            match feedback_rx.try_next() {
+                Ok(Some(response)) => responses.push(response),
+                Err(_) | Ok(None) => {
+                    error!("somehow we received fewer feedback responses than sent messages")
+                }
+            }
+        }
+
+        // the channel should have been drained and all sender should have been dropped
+        debug_assert!(matches!(feedback_rx.try_next(), Ok(None)));
+        responses
+    }
+
+    pub(crate) async fn broadcast(&self, msg: OffchainMessage) {
+        if self.addresses.is_empty() {
+            warn!("attempting to broadcast {} while no remotes are known", msg);
+            return;
+        }
+
+        debug!("broadcasting {} to {} remotes", msg, self.addresses.len());
+        stream::iter(self.create_broadcast_configs(msg, None))
+            .for_each_concurrent(self.concurrency_level, |cfg| cfg.send())
+            .await
+    }
+}
+
+// internal struct to have per-connection config on hand
+struct BroadcastConfig {
+    address: SocketAddr,
+    config: ConnectionConfig,
+    feedback_sender: Option<FeedbackSender>,
+    message: OffchainMessage,
+}
+
+impl BroadcastConfig {
+    async fn send(self) {
+        let response = send_single_message(self.address, self.config, &self.message).await;
+        if let Some(feedback_sender) = self.feedback_sender {
+            // this can only fail if the receiver is disconnected which should never be the case
+            // thus we can ignore the possible error
+            let _ = feedback_sender.unbounded_send(response);
+        } else if let Err(err) = response.response {
+            // if we're not forwarding feedback, at least emit a warning about the failure
+            warn!(
+                "failed to broadcast {} to {} - {}",
+                self.message, self.address, err
+            )
+        }
+    }
+}

validator-api/src/networking/sender/ephemeral.rs

@@ -0,0 +1,83 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::codec::OffchainCodec;
+use crate::networking::error::NetworkingError;
+use crate::networking::message::OffchainMessage;
+use crate::networking::sender::ConnectionConfig;
+use futures::{SinkExt, StreamExt};
+use std::io;
+use std::net::SocketAddr;
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::time::timeout;
+use tokio_util::codec::Framed;
+
+// this connection only exists for a single message
+pub(crate) struct EphemeralConnection {
+    remote: SocketAddr,
+    conn: Framed<TcpStream, OffchainCodec>,
+}
+
+impl EphemeralConnection {
+    pub(crate) async fn connect(
+        address: SocketAddr,
+        connection_timeout: Duration,
+    ) -> io::Result<Self> {
+        trace!("attempting to connect to {}", address);
+        let conn = match timeout(connection_timeout, TcpStream::connect(address)).await {
+            Err(_timeout) => {
+                return Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("timed out while attempting to send message to {}", address),
+                ))
+            }
+            Ok(conn_res) => conn_res?,
+        };
+        let framed_conn = Framed::new(conn, OffchainCodec);
+        Ok(Self {
+            remote: address,
+            conn: framed_conn,
+        })
+    }
+
+    pub(crate) fn remote(&self) -> SocketAddr {
+        self.remote
+    }
+
+    async fn send(
+        &mut self,
+        message: &OffchainMessage,
+        send_timeout: Duration,
+        response_timeout: Option<Duration>,
+    ) -> Result<Option<OffchainMessage>, NetworkingError> {
+        trace!("attempting to send to {}", self.remote);
+        match timeout(send_timeout, self.conn.send(message)).await {
+            Err(_timeout) => {
+                return Err(NetworkingError::Io(io::Error::new(
+                    io::ErrorKind::Other,
+                    "timed out while attempting to send message",
+                )))
+            }
+            Ok(res) => res?,
+        }
+        if let Some(response_timeout) = response_timeout {
+            match timeout(response_timeout, self.conn.next()).await {
+                Err(_elapsed) => Ok(None),
+                Ok(response) => response.transpose(),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub(crate) async fn connect_and_send(
+        address: SocketAddr,
+        cfg: ConnectionConfig,
+        message: &OffchainMessage,
+    ) -> Result<Option<OffchainMessage>, NetworkingError> {
+        let mut conn = EphemeralConnection::connect(address, cfg.connection_timeout).await?;
+        conn.send(message, cfg.send_timeout, cfg.response_timeout)
+            .await
+    }
+}

validator-api/src/networking/sender/mod.rs

@@ -0,0 +1,65 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::networking::error::NetworkingError;
+use crate::networking::message::OffchainMessage;
+use crate::networking::sender::broadcast::Broadcaster;
+use crate::networking::sender::ephemeral::EphemeralConnection;
+use std::net::SocketAddr;
+use std::time::Duration;
+
+pub(crate) mod broadcast;
+pub(crate) mod ephemeral;
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct ConnectionConfig {
+    pub(crate) connection_timeout: Duration,
+    pub(crate) response_timeout: Option<Duration>,
+    pub(crate) send_timeout: Duration,
+}
+
+pub(crate) struct SendResponse {
+    pub(crate) source: SocketAddr,
+    pub(crate) response: Result<Option<OffchainMessage>, NetworkingError>,
+}
+
+impl SendResponse {
+    pub(crate) fn new(
+        source: SocketAddr,
+        response: Result<Option<OffchainMessage>, NetworkingError>,
+    ) -> Self {
+        SendResponse { source, response }
+    }
+}
+
+pub(crate) async fn send_single_message(
+    address: SocketAddr,
+    cfg: ConnectionConfig,
+    message: &OffchainMessage,
+) -> SendResponse {
+    let res = EphemeralConnection::connect_and_send(address, cfg, message).await;
+    SendResponse::new(address, res)
+}
+
+pub(crate) async fn broadcast_message(
+    addresses: Vec<SocketAddr>,
+    cfg: ConnectionConfig,
+    message: &OffchainMessage,
+) {
+    Broadcaster::new(addresses, cfg)
+        .broadcast(message.clone())
+        .await
+}
+
+pub(crate) async fn broadcast_message_with_feedback(
+    addresses: Vec<SocketAddr>,
+    cfg: ConnectionConfig,
+    message: &OffchainMessage,
+) -> Vec<SendResponse> {
+    Broadcaster::new(addresses, cfg)
+        .broadcast_with_feedback(message.clone())
+        .await
+}
+
+// NOTE: for the original purposes of DKG stateless broadcasts and one-off sends were enough,
+// so I never implemented proper persistent connections