TEMP gh runner fix

* using common middleware for all http servers

* improved span handling in credential-proxy

* ensure increase in sequence number upon making deposit

* added explicit connect options for the db

* fixed further instances of incorrect span instrumentation

* batch deposit requests together to improve concurrency

* ignore cancelled requests

* updated credential proxy version to 0.1.4

* adjusted Dockerfile with new binary location

* log binary version on startup

* reduce default log level

* guard against unavaiable commit sha

* apply review comments: dont exit(0), instead just shutdown normally

* add skip_webhook parameter to obtain-async

* removing dead code

.github/workflows/ci-build-upload-binaries.yml

@@ -21,7 +21,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        platform: [ arc-ubuntu-20.04 ]
+        platform: [ arc-ubuntu-22.04 ]
 
     runs-on: ${{ matrix.platform }}
     env:

CHANGELOG.md

@@ -4,6 +4,10 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ## [Unreleased]
 
+## [2024.13-magura-drift] (2024-11-29)
+
+- Optimised syncing bandwidth information to storage
+
 ## [2024.13-magura-patched] (2024-11-22)
 
 - [experimental] allow clients to change between deterministic route selection based on packet headers and a pseudorandom distribution

Cargo.lock

@@ -6890,6 +6890,7 @@ dependencies = [
  "nym-task",
  "nym-wireguard-types",
  "thiserror",
+ "time",
  "tokio",
  "tokio-stream",
  "x25519-dalek",

common/client-core/src/client/topology_control/accessor.rs

@@ -38,7 +38,7 @@ pub struct TopologyReadPermit<'a> {
     permit: RwLockReadGuard<'a, Option<NymTopology>>,
 }
 
-impl<'a> Deref for TopologyReadPermit<'a> {
+impl Deref for TopologyReadPermit<'_> {
     type Target = Option<NymTopology>;
 
     fn deref(&self) -> &Self::Target {

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

@@ -32,7 +32,7 @@ impl Div<GasPrice> for Coin {
     }
 }
 
-impl<'a> Div<GasPrice> for &'a Coin {
+impl Div<GasPrice> for &Coin {
     type Output = Gas;
 
     fn div(self, rhs: GasPrice) -> Self::Output {

common/client-libs/validator-client/src/nyxd/fee/gas_price.rs

@@ -22,7 +22,7 @@ pub struct GasPrice {
     pub denom: String,
 }
 
-impl<'a> Mul<Gas> for &'a GasPrice {
+impl Mul<Gas> for &GasPrice {
     type Output = Coin;
 
     fn mul(self, gas_limit: Gas) -> Self::Output {

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

@@ -32,7 +32,7 @@ pub(crate) mod string_rfc3339_offset_date_time {
 
     struct Rfc3339OffsetDateTimeVisitor;
 
-    impl<'de> Visitor<'de> for Rfc3339OffsetDateTimeVisitor {
+    impl Visitor<'_> for Rfc3339OffsetDateTimeVisitor {
         type Value = OffsetDateTime;
 
         fn expecting(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {

common/credential-verification/src/bandwidth_storage_manager.rs

@@ -111,7 +111,7 @@ impl<S: Storage + Clone + 'static> BandwidthStorageManager<S> {
     }
 
     #[instrument(level = "trace", skip_all)]
-    async fn sync_storage_bandwidth(&mut self) -> Result<()> {
+    pub async fn sync_storage_bandwidth(&mut self) -> Result<()> {
         trace!("syncing client bandwidth with the underlying storage");
         let updated = self
             .storage

common/credential-verification/src/client_bandwidth.rs

@@ -8,8 +8,8 @@ use std::time::Duration;
 use time::OffsetDateTime;
 use tokio::sync::RwLock;
 
-const DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE: Duration = Duration::from_millis(5);
-const DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT: i64 = 512 * 1024; // 512kB
+const DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE: Duration = Duration::from_secs(5 * 60); // 5 minutes
+const DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT: i64 = 5 * 1024 * 1024; // 5MB
 
 #[derive(Debug, Clone, Copy)]
 pub struct BandwidthFlushingBehaviourConfig {

common/dkg/src/bte/proof_chunking.rs

@@ -26,9 +26,8 @@ const PARALLEL_RUNS: usize = 32;
 /// `lambda` ($\lambda$) in the DKG paper
 const SECURITY_PARAMETER: usize = 256;
 
-// note: ceiling in integer division can be achieved via q = (x + y - 1) / y;
 /// ceil(SECURITY_PARAMETER / PARALLEL_RUNS) in the paper
-const NUM_CHALLENGE_BITS: usize = (SECURITY_PARAMETER + PARALLEL_RUNS - 1) / PARALLEL_RUNS;
+const NUM_CHALLENGE_BITS: usize = SECURITY_PARAMETER.div_ceil(PARALLEL_RUNS);
 
 // type alias for ease of use
 type FirstChallenge = Vec<Vec<Vec<u64>>>;

common/dkg/src/interpolation/polynomial.rs

@@ -196,7 +196,7 @@ impl<'b> Add<&'b Polynomial> for Polynomial {
     }
 }
 
-impl<'a> Add<Polynomial> for &'a Polynomial {
+impl Add<Polynomial> for &Polynomial {
     type Output = Polynomial;
 
     fn add(self, rhs: Polynomial) -> Polynomial {
@@ -212,10 +212,10 @@ impl Add<Polynomial> for Polynomial {
     }
 }
 
-impl<'a, 'b> Add<&'b Polynomial> for &'a Polynomial {
+impl<'a> Add<&'a Polynomial> for &Polynomial {
     type Output = Polynomial;
 
-    fn add(self, rhs: &'b Polynomial) -> Self::Output {
+    fn add(self, rhs: &'a Polynomial) -> Self::Output {
         let len = self.coefficients.len();
         let rhs_len = rhs.coefficients.len();
 

common/gateway-requests/src/registration/handshake/mod.rs

@@ -37,7 +37,7 @@ pub struct GatewayHandshake<'a> {
     handshake_future: BoxFuture<'a, Result<SharedGatewayKey, HandshakeError>>,
 }
 
-impl<'a> Future for GatewayHandshake<'a> {
+impl Future for GatewayHandshake<'_> {
     type Output = Result<SharedGatewayKey, HandshakeError>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

common/nym_offline_compact_ecash/src/scheme/coin_indices_signatures.rs

@@ -324,18 +324,6 @@ pub fn unchecked_aggregate_indices_signatures(
     _aggregate_indices_signatures(params, vk, signatures_shares, false)
 }
 
-/// Generates parameters for the scheme setup.
-///
-/// # Arguments
-///
-/// * `total_coins` - it is the number of coins in a freshly generated wallet. It is the public parameter of the scheme.
-///
-/// # Returns
-///
-/// A `Parameters` struct containing group parameters, public key, the number of signatures (`total_coins`),
-/// and a map of signatures for each index `l`.
-///
-
 #[cfg(test)]
 mod tests {
     use super::*;

common/nym_offline_compact_ecash/src/scheme/keygen.rs

@@ -264,7 +264,7 @@ impl<'b> Add<&'b VerificationKeyAuth> for VerificationKeyAuth {
     }
 }
 
-impl<'a> Mul<Scalar> for &'a VerificationKeyAuth {
+impl Mul<Scalar> for &VerificationKeyAuth {
     type Output = VerificationKeyAuth;
 
     #[inline]

common/nym_offline_compact_ecash/src/scheme/mod.rs

@@ -984,7 +984,7 @@ pub struct SerialNumberRef<'a> {
     pub(crate) inner: &'a [G1Projective],
 }
 
-impl<'a> SerialNumberRef<'a> {
+impl SerialNumberRef<'_> {
     pub fn to_bytes(&self) -> Vec<u8> {
         let ss_len = self.inner.len();
         let mut bytes: Vec<u8> = Vec::with_capacity(ss_len * 48);

common/nymcoconut/src/elgamal.rs

@@ -206,10 +206,10 @@ impl Deref for PublicKey {
     }
 }
 
-impl<'a, 'b> Mul<&'b Scalar> for &'a PublicKey {
+impl<'a> Mul<&'a Scalar> for &PublicKey {
     type Output = G1Projective;
 
-    fn mul(self, rhs: &'b Scalar) -> Self::Output {
+    fn mul(self, rhs: &'a Scalar) -> Self::Output {
         self.0 * rhs
     }
 }

common/nymcoconut/src/scheme/keygen.rs

@@ -305,7 +305,7 @@ impl<'b> Add<&'b VerificationKey> for VerificationKey {
     }
 }
 
-impl<'a> Mul<Scalar> for &'a VerificationKey {
+impl Mul<Scalar> for &VerificationKey {
     type Output = VerificationKey;
 
     #[inline]

common/nymsphinx/addressing/src/clients.rs

@@ -64,7 +64,7 @@ impl<'de> Deserialize<'de> for Recipient {
     {
         struct RecipientVisitor;
 
-        impl<'de> Visitor<'de> for RecipientVisitor {
+        impl Visitor<'_> for RecipientVisitor {
             type Value = Recipient;
 
             fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {

common/nymsphinx/addressing/src/nodes.rs

@@ -1,6 +1,12 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+//! Encodoing and decoding node routing information.
+//!
+//! This module is responsible for encoding and decoding node routing information, so that
+//! they could be later put into an appropriate field in a sphinx header.
+//! Currently, that routing information is an IP address, but in principle it can be anything
+//!  for as long as it's going to fit in the field.
 use nym_crypto::asymmetric::identity;
 use nym_sphinx_types::{NodeAddressBytes, NODE_ADDRESS_LENGTH};
 
@@ -12,13 +18,6 @@ use thiserror::Error;
 pub type NodeIdentity = identity::PublicKey;
 pub const NODE_IDENTITY_SIZE: usize = identity::PUBLIC_KEY_LENGTH;
 
-/// Encodoing and decoding node routing information.
-///
-/// This module is responsible for encoding and decoding node routing information, so that
-/// they could be later put into an appropriate field in a sphinx header.
-/// Currently, that routing information is an IP address, but in principle it can be anything
-/// for as long as it's going to fit in the field.
-
 /// MAX_UNPADDED_LEN represents maximum length an unpadded address could have.
 /// In this case it's an ipv6 socket address (with version prefix)
 pub const MAX_NODE_ADDRESS_UNPADDED_LEN: usize = 19;

common/nymsphinx/anonymous-replies/src/reply_surb.rs

@@ -56,7 +56,7 @@ impl<'de> Deserialize<'de> for ReplySurb {
     {
         struct ReplySurbVisitor;
 
-        impl<'de> Visitor<'de> for ReplySurbVisitor {
+        impl Visitor<'_> for ReplySurbVisitor {
             type Value = ReplySurb;
 
             fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {

common/socks5/requests/src/request.rs

@@ -253,25 +253,15 @@ impl Socks5RequestContent {
     /// Deserialize the request type, connection id, destination address and port,
     /// and the request body from bytes.
     ///
-    // TODO: this was already inaccurate
-    // /// Serialized bytes looks like this:
-    // ///
-    // /// --------------------------------------------------------------------------------------
-    // ///  request_flag | connection_id | address_length | remote_address_bytes | request_data |
-    // ///        1      |       8       |      2         |    address_length    |    ...       |
-    // /// --------------------------------------------------------------------------------------
-    ///
     /// The request_flag tells us whether this is a new connection request (`new_connect`),
     /// an already-established connection we should send up (`new_send`), or
     /// a request to close an established connection (`new_close`).
-
     // connect:
     // RequestFlag::Connect || CONN_ID || ADDR_LEN || ADDR || <RETURN_ADDR>
     //
     // send:
     // RequestFlag::Send || CONN_ID || LOCAL_CLOSED || DATA
     // where DATA: SEQ || TRUE_DATA
-
     pub fn try_from_bytes(b: &[u8]) -> Result<Socks5RequestContent, RequestDeserializationError> {
         // each request needs to at least contain flag and ConnectionId
         if b.is_empty() {

common/wireguard/Cargo.toml

@@ -26,6 +26,7 @@ log.workspace = true
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] }
 tokio-stream = { workspace = true }
+time = { workspace = true }
 
 nym-authenticator-requests = { path = "../authenticator-requests" }
 nym-credential-verification = { path = "../credential-verification" }

common/wireguard/src/lib.rs

@@ -20,6 +20,7 @@ use tokio::sync::mpsc::{self, Receiver, Sender};
 pub(crate) mod error;
 pub mod peer_controller;
 pub mod peer_handle;
+pub mod peer_storage_manager;
 
 pub struct WgApiWrapper {
     inner: WGApi,
@@ -118,7 +119,7 @@ pub async fn start_wireguard<St: nym_gateway_storage::Storage + Clone + 'static>
         storage
             .insert_wireguard_peer(peer, bandwidth_manager.is_some())
             .await?;
-        peer_bandwidth_managers.insert(peer.public_key.clone(), bandwidth_manager);
+        peer_bandwidth_managers.insert(peer.public_key.clone(), (bandwidth_manager, peer.clone()));
     }
     wg_api.create_interface()?;
     let interface_config = InterfaceConfiguration {

common/wireguard/src/peer_controller.rs

@@ -20,9 +20,9 @@ use std::{collections::HashMap, sync::Arc};
 use tokio::sync::{mpsc, RwLock};
 use tokio_stream::{wrappers::IntervalStream, StreamExt};
 
-use crate::peer_handle::PeerHandle;
 use crate::WgApiWrapper;
 use crate::{error::Error, peer_handle::SharedBandwidthStorageManager};
+use crate::{peer_handle::PeerHandle, peer_storage_manager::PeerStorageManager};
 
 pub enum PeerControlRequest {
     AddPeer {
@@ -79,7 +79,7 @@ impl<St: Storage + Clone + 'static> PeerController<St> {
         storage: St,
         wg_api: Arc<WgApiWrapper>,
         initial_host_information: Host,
-        bw_storage_managers: HashMap<Key, Option<SharedBandwidthStorageManager<St>>>,
+        bw_storage_managers: HashMap<Key, (Option<SharedBandwidthStorageManager<St>>, Peer)>,
         request_tx: mpsc::Sender<PeerControlRequest>,
         request_rx: mpsc::Receiver<PeerControlRequest>,
         task_client: nym_task::TaskClient,
@@ -88,11 +88,16 @@ impl<St: Storage + Clone + 'static> PeerController<St> {
             tokio::time::interval(DEFAULT_PEER_TIMEOUT_CHECK),
         );
         let host_information = Arc::new(RwLock::new(initial_host_information));
-        for (public_key, bandwidth_storage_manager) in bw_storage_managers.iter() {
-            let mut handle = PeerHandle::new(
+        for (public_key, (bandwidth_storage_manager, peer)) in bw_storage_managers.iter() {
+            let peer_storage_manager = PeerStorageManager::new(
                 storage.clone(),
+                peer.clone(),
+                bandwidth_storage_manager.is_some(),
+            );
+            let mut handle = PeerHandle::new(
                 public_key.clone(),
                 host_information.clone(),
+                peer_storage_manager,
                 bandwidth_storage_manager.clone(),
                 request_tx.clone(),
                 &task_client,
@@ -103,6 +108,10 @@ impl<St: Storage + Clone + 'static> PeerController<St> {
                 }
             });
         }
+        let bw_storage_managers = bw_storage_managers
+            .into_iter()
+            .map(|(k, (m, _))| (k, m))
+            .collect();
 
         PeerController {
             storage,
@@ -184,10 +193,15 @@ impl<St: Storage + Clone + 'static> PeerController<St> {
             Self::generate_bandwidth_manager(self.storage.clone(), &peer.public_key)
                 .await?
                 .map(|bw_m| Arc::new(RwLock::new(bw_m)));
-        let mut handle = PeerHandle::new(
+        let peer_storage_manager = PeerStorageManager::new(
             self.storage.clone(),
+            peer.clone(),
+            bandwidth_storage_manager.is_some(),
+        );
+        let mut handle = PeerHandle::new(
             peer.public_key.clone(),
             self.host_information.clone(),
+            peer_storage_manager,
             bandwidth_storage_manager.clone(),
             self.request_tx.clone(),
             &self.task_client,

common/wireguard/src/peer_handle.rs

@@ -3,6 +3,7 @@
 
 use crate::error::Error;
 use crate::peer_controller::PeerControlRequest;
+use crate::peer_storage_manager::PeerStorageManager;
 use defguard_wireguard_rs::host::Peer;
 use defguard_wireguard_rs::{host::Host, key::Key};
 use futures::channel::oneshot;
@@ -21,9 +22,9 @@ pub(crate) type SharedBandwidthStorageManager<St> = Arc<RwLock<BandwidthStorageM
 const AUTO_REMOVE_AFTER: Duration = Duration::from_secs(60 * 60 * 24 * 30); // 30 days
 
 pub struct PeerHandle<St> {
-    storage: St,
     public_key: Key,
     host_information: Arc<RwLock<Host>>,
+    peer_storage_manager: PeerStorageManager<St>,
     bandwidth_storage_manager: Option<SharedBandwidthStorageManager<St>>,
     request_tx: mpsc::Sender<PeerControlRequest>,
     timeout_check_interval: IntervalStream,
@@ -33,9 +34,9 @@ pub struct PeerHandle<St> {
 
 impl<St: Storage + Clone + 'static> PeerHandle<St> {
     pub fn new(
-        storage: St,
         public_key: Key,
         host_information: Arc<RwLock<Host>>,
+        peer_storage_manager: PeerStorageManager<St>,
         bandwidth_storage_manager: Option<SharedBandwidthStorageManager<St>>,
         request_tx: mpsc::Sender<PeerControlRequest>,
         task_client: &TaskClient,
@@ -46,9 +47,9 @@ impl<St: Storage + Clone + 'static> PeerHandle<St> {
         let mut task_client = task_client.fork(format!("peer-{public_key}"));
         task_client.disarm();
         PeerHandle {
-            storage,
             public_key,
             host_information,
+            peer_storage_manager,
             bandwidth_storage_manager,
             request_tx,
             timeout_check_interval,
@@ -84,16 +85,19 @@ impl<St: Storage + Clone + 'static> PeerHandle<St> {
                 .ok_or(Error::InconsistentConsumedBytes)?
                 .try_into()
                 .map_err(|_| Error::InconsistentConsumedBytes)?;
-            if spent_bandwidth > 0
-                && bandwidth_manager
+            if spent_bandwidth > 0 {
+                self.peer_storage_manager.update_trx(kernel_peer);
+                if bandwidth_manager
                     .write()
                     .await
                     .try_use_bandwidth(spent_bandwidth)
                     .await
                     .is_err()
-            {
-                let success = self.remove_peer().await?;
-                return Ok(!success);
+                {
+                    let success = self.remove_peer().await?;
+                    self.peer_storage_manager.remove_peer();
+                    return Ok(!success);
+                }
             }
         } else {
             if SystemTime::now().duration_since(self.startup_timestamp)? >= AUTO_REMOVE_AFTER {
@@ -132,7 +136,7 @@ impl<St: Storage + Clone + 'static> PeerHandle<St> {
                             // the host information hasn't beed updated yet
                             continue;
                         };
-                    let Some(storage_peer) = self.storage.get_wireguard_peer(&self.public_key.to_string()).await? else {
+                    let Some(storage_peer) = self.peer_storage_manager.get_peer() else {
                         log::debug!("Peer {:?} not in storage anymore, shutting down handle", self.public_key);
                         return Ok(());
                     };
@@ -141,12 +145,18 @@ impl<St: Storage + Clone + 'static> PeerHandle<St> {
                         return Ok(());
                     } else {
                         // Update storage values
-                        self.storage.insert_wireguard_peer(&kernel_peer, self.bandwidth_storage_manager.is_some()).await?;
+                        self.peer_storage_manager.sync_storage_peer().await?;
                     }
                 }
 
                 _ = self.task_client.recv() => {
                     log::trace!("PeerHandle: Received shutdown");
+                    if let Some(bandwidth_manager) = &self.bandwidth_storage_manager {
+                        if let Err(e) = bandwidth_manager.write().await.sync_storage_bandwidth().await {
+                            log::error!("Storage sync failed - {e}, unaccounted bandwidth might have been consumed");
+                        }
+                    }
+                    log::trace!("PeerHandle: Finished shutdown");
                 }
             }
         }

common/wireguard/src/peer_storage_manager.rs

@@ -0,0 +1,138 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::Error;
+use defguard_wireguard_rs::host::Peer;
+use nym_gateway_storage::models::WireguardPeer;
+use nym_gateway_storage::Storage;
+use std::time::Duration;
+use time::OffsetDateTime;
+
+const DEFAULT_PEER_MAX_FLUSHING_RATE: Duration = Duration::from_secs(60 * 60 * 24); // 24h
+const DEFAULT_PEER_MAX_DELTA_FLUSHING_AMOUNT: u64 = 512 * 1024 * 1024; // 512MB
+
+#[derive(Debug, Clone, Copy)]
+pub struct PeerFlushingBehaviourConfig {
+    /// Defines maximum delay between peer information being flushed to the persistent storage.
+    pub peer_max_flushing_rate: Duration,
+
+    /// Defines a maximum change in peer before it gets flushed to the persistent storage.
+    pub peer_max_delta_flushing_amount: u64,
+}
+
+impl Default for PeerFlushingBehaviourConfig {
+    fn default() -> Self {
+        Self {
+            peer_max_flushing_rate: DEFAULT_PEER_MAX_FLUSHING_RATE,
+            peer_max_delta_flushing_amount: DEFAULT_PEER_MAX_DELTA_FLUSHING_AMOUNT,
+        }
+    }
+}
+
+pub struct PeerStorageManager<S> {
+    pub(crate) storage: S,
+    pub(crate) peer_information: Option<PeerInformation>,
+    pub(crate) cfg: PeerFlushingBehaviourConfig,
+    pub(crate) with_client_id: bool,
+}
+
+impl<S: Storage + Clone + 'static> PeerStorageManager<S> {
+    pub(crate) fn new(storage: S, peer: Peer, with_client_id: bool) -> Self {
+        let peer_information = Some(PeerInformation::new(peer));
+        Self {
+            storage,
+            peer_information,
+            cfg: PeerFlushingBehaviourConfig::default(),
+            with_client_id,
+        }
+    }
+
+    pub(crate) fn get_peer(&self) -> Option<WireguardPeer> {
+        self.peer_information
+            .as_ref()
+            .map(|p| p.peer.clone().into())
+    }
+
+    pub(crate) fn remove_peer(&mut self) {
+        self.peer_information = None;
+    }
+
+    pub(crate) fn update_trx(&mut self, kernel_peer: &Peer) {
+        if let Some(peer_information) = self.peer_information.as_mut() {
+            peer_information.update_trx_bytes(kernel_peer.tx_bytes, kernel_peer.rx_bytes);
+        }
+    }
+
+    pub(crate) async fn sync_storage_peer(&mut self) -> Result<(), Error> {
+        let Some(peer_information) = self.peer_information.as_mut() else {
+            return Ok(());
+        };
+        if !peer_information.should_sync(self.cfg) {
+            return Ok(());
+        }
+        if self
+            .storage
+            .get_wireguard_peer(&peer_information.peer().public_key.to_string())
+            .await?
+            .is_none()
+        {
+            self.peer_information = None;
+            return Ok(());
+        }
+        self.storage
+            .insert_wireguard_peer(peer_information.peer(), self.with_client_id)
+            .await?;
+
+        peer_information.resync_peer_with_storage();
+
+        Ok(())
+    }
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct PeerInformation {
+    pub(crate) peer: Peer,
+    pub(crate) last_synced: OffsetDateTime,
+
+    pub(crate) bytes_delta_since_sync: u64,
+}
+
+impl PeerInformation {
+    pub fn new(peer: Peer) -> PeerInformation {
+        PeerInformation {
+            peer,
+            last_synced: OffsetDateTime::now_utc(),
+            bytes_delta_since_sync: 0,
+        }
+    }
+
+    pub(crate) fn should_sync(&self, cfg: PeerFlushingBehaviourConfig) -> bool {
+        if self.bytes_delta_since_sync >= cfg.peer_max_delta_flushing_amount {
+            return true;
+        }
+
+        if self.last_synced + cfg.peer_max_flushing_rate < OffsetDateTime::now_utc()
+            && self.bytes_delta_since_sync != 0
+        {
+            return true;
+        }
+
+        false
+    }
+
+    pub(crate) fn peer(&self) -> &Peer {
+        &self.peer
+    }
+
+    pub(crate) fn update_trx_bytes(&mut self, tx_bytes: u64, rx_bytes: u64) {
+        self.bytes_delta_since_sync += tx_bytes.saturating_sub(self.peer.tx_bytes)
+            + rx_bytes.saturating_sub(self.peer.rx_bytes);
+        self.peer.tx_bytes = tx_bytes;
+        self.peer.rx_bytes = rx_bytes;
+    }
+
+    pub(crate) fn resync_peer_with_storage(&mut self) {
+        self.bytes_delta_since_sync = 0;
+        self.last_synced = OffsetDateTime::now_utc();
+    }
+}

explorer-api/src/geo_ip/location.rs

@@ -166,7 +166,7 @@ impl GeoIp {
     }
 }
 
-impl<'a> TryFrom<&City<'a>> for Location {
+impl TryFrom<&City<'_>> for Location {
     type Error = String;
 
     fn try_from(city: &City) -> Result<Self, Self::Error> {

explorer-api/src/guards/location.rs

@@ -65,7 +65,7 @@ impl<'r> FromRequest<'r> for Location {
     }
 }
 
-impl<'a> OpenApiFromRequest<'a> for Location {
+impl OpenApiFromRequest<'_> for Location {
     fn from_request_input(
         _gen: &mut OpenApiGenerator,
         _name: String,

gateway/src/node/helpers.rs

@@ -5,6 +5,8 @@ use async_trait::async_trait;
 use nym_sdk::{NymApiTopologyProvider, NymApiTopologyProviderConfig, UserAgent};
 use nym_topology::{gateway, NymTopology, TopologyProvider};
 use std::sync::Arc;
+use std::time::Duration;
+use time::OffsetDateTime;
 use tokio::sync::Mutex;
 use tracing::debug;
 use url::Url;
@@ -17,6 +19,7 @@ pub struct GatewayTopologyProvider {
 impl GatewayTopologyProvider {
     pub fn new(
         gateway_node: gateway::LegacyNode,
+        cache_ttl: Duration,
         user_agent: UserAgent,
         nym_api_url: Vec<Url>,
     ) -> GatewayTopologyProvider {
@@ -31,6 +34,9 @@ impl GatewayTopologyProvider {
                     env!("CARGO_PKG_VERSION").to_string(),
                     Some(user_agent),
                 ),
+                cache_ttl,
+                cached_at: OffsetDateTime::UNIX_EPOCH,
+                cached: None,
                 gateway_node,
             })),
         }
@@ -39,25 +45,53 @@ impl GatewayTopologyProvider {
 
 struct GatewayTopologyProviderInner {
     inner: NymApiTopologyProvider,
+    cache_ttl: Duration,
+    cached_at: OffsetDateTime,
+    cached: Option<NymTopology>,
     gateway_node: gateway::LegacyNode,
 }
 
+impl GatewayTopologyProviderInner {
+    fn cached_topology(&self) -> Option<NymTopology> {
+        if let Some(cached_topology) = &self.cached {
+            if self.cached_at + self.cache_ttl > OffsetDateTime::now_utc() {
+                return Some(cached_topology.clone());
+            }
+        }
+
+        None
+    }
+
+    async fn update_cache(&mut self) -> Option<NymTopology> {
+        let updated_cache = match self.inner.get_new_topology().await {
+            None => None,
+            Some(mut base) => {
+                if !base.gateway_exists(&self.gateway_node.identity_key) {
+                    debug!(
+                        "{} didn't exist in topology. inserting it.",
+                        self.gateway_node.identity_key
+                    );
+                    base.insert_gateway(self.gateway_node.clone());
+                }
+                Some(base)
+            }
+        };
+
+        self.cached_at = OffsetDateTime::now_utc();
+        self.cached = updated_cache.clone();
+
+        updated_cache
+    }
+}
+
 #[async_trait]
 impl TopologyProvider for GatewayTopologyProvider {
     async fn get_new_topology(&mut self) -> Option<NymTopology> {
         let mut guard = self.inner.lock().await;
-        match guard.inner.get_new_topology().await {
-            None => None,
-            Some(mut base) => {
-                if !base.gateway_exists(&guard.gateway_node.identity_key) {
-                    debug!(
-                        "{} didn't exist in topology. inserting it.",
-                        guard.gateway_node.identity_key
-                    );
-                    base.insert_gateway(guard.gateway_node.clone());
-                }
-                Some(base)
-            }
+        // check the cache
+        if let Some(cached) = guard.cached_topology() {
+            return Some(cached);
         }
+        guard.update_cache().await
     }
 }

gateway/src/node/mod.rs

@@ -30,6 +30,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::path::PathBuf;
 use std::process;
 use std::sync::Arc;
+use std::time::Duration;
 use tracing::*;
 
 pub(crate) mod client_handling;
@@ -148,8 +149,11 @@ impl<St> Gateway<St> {
     }
 
     fn gateway_topology_provider(&self) -> GatewayTopologyProvider {
+        // TODO: make topology ttl configurable
+        // (to be done in reeses with the final smooshing)
         GatewayTopologyProvider::new(
             self.as_topology_node(),
+            Duration::from_secs(5 * 60),
             self.user_agent.clone(),
             self.config.gateway.nym_api_urls.clone(),
         )

nym-api/nym-api-requests/src/helpers.rs

@@ -64,7 +64,7 @@ pub(crate) mod overengineered_offset_date_time_serde {
         ])),
     ];
 
-    impl<'de> Visitor<'de> for OffsetDateTimeVisitor {
+    impl Visitor<'_> for OffsetDateTimeVisitor {
         type Value = OffsetDateTime;
 
         fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result {

nym-api/src/ecash/tests/helpers.rs

@@ -76,7 +76,7 @@ pub(crate) async fn submit_public_keys(controllers: &mut [TestingDkgController],
             .unwrap();
     }
 
-    let threshold = (2 * controllers.len() as u64 + 3 - 1) / 3;
+    let threshold = (2 * controllers.len() as u64).div_ceil(3);
 
     let mut guard = controllers[0].chain_state.lock().unwrap();
     guard.dkg_contract.epoch.state = EpochState::DealingExchange { resharing };

nym-api/src/network_monitor/monitor/gateway_clients_cache.rs

@@ -59,7 +59,7 @@ impl GatewayClientHandle {
     }
 }
 
-impl<'a> UnlockedGatewayClientHandle<'a> {
+impl UnlockedGatewayClientHandle<'_> {
     pub(crate) fn get_mut_unchecked(
         &mut self,
     ) -> &mut GatewayClient<nyxd::Client, PersistentStorage> {

nym-credential-proxy/nym-credential-proxy/src/deposit_maker.rs

@@ -7,7 +7,6 @@ use nym_crypto::asymmetric::ed25519;
 use nym_ecash_contract_common::deposit::DepositId;
 use nym_validator_client::nyxd::cosmwasm_client::ContractResponseData;
 use nym_validator_client::nyxd::{Coin, Hash};
-use std::process;
 use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};

nym-credential-proxy/nym-credential-proxy/src/http/state/mod.rs

@@ -759,7 +759,7 @@ impl<'a> ChainWritePermit<'a> {
     }
 }
 
-impl<'a> Deref for ChainWritePermit<'a> {
+impl Deref for ChainWritePermit<'_> {
     type Target = DirectSigningHttpRpcNyxdClient;
 
     fn deref(&self) -> &Self::Target {

sdk/ffi/go/src/lib.rs

@@ -1,6 +1,9 @@
 // Copyright 2023-2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+// due to autogenerated code
+#![allow(clippy::empty_line_after_doc_comments)]
+
 use nym_sdk::mixnet::Recipient;
 use nym_sphinx_anonymous_replies::requests::AnonymousSenderTag;
 uniffi::include_scaffolding!("bindings");

service-providers/authenticator/src/mixnet_listener.rs

@@ -316,6 +316,7 @@ impl<S: Storage + Clone + 'static> MixnetListener<S> {
             .filter(|r| r.1.is_none())
             .choose(&mut thread_rng())
             .ok_or(AuthenticatorError::NoFreeIp)?;
+        let private_ips = *private_ip_ref.0;
         // mark it as used, even though it's not final
         *private_ip_ref.1 = Some(SystemTime::now());
         let gateway_data = GatewayClient::new(
@@ -337,11 +338,12 @@ impl<S: Storage + Clone + 'static> MixnetListener<S> {
                 v1::response::AuthenticatorResponse::new_pending_registration_success(
                     v1::registration::RegistrationData {
                         nonce: registration_data.nonce,
-                        gateway_data: v1::GatewayClient {
-                            pub_key: gateway_data.pub_key,
-                            private_ip: gateway_data.private_ips.ipv4.into(),
-                            mac: v1::ClientMac::new(gateway_data.mac.to_vec()),
-                        },
+                        gateway_data: v1::registration::GatewayClient::new(
+                            self.keypair().private_key(),
+                            remote_public.inner(),
+                            private_ips.ipv4.into(),
+                            nonce,
+                        ),
                         wg_port: registration_data.wg_port,
                     },
                     request_id,
@@ -356,7 +358,12 @@ impl<S: Storage + Clone + 'static> MixnetListener<S> {
                 v2::response::AuthenticatorResponse::new_pending_registration_success(
                     v2::registration::RegistrationData {
                         nonce: registration_data.nonce,
-                        gateway_data: registration_data.gateway_data.into(),
+                        gateway_data: v2::registration::GatewayClient::new(
+                            self.keypair().private_key(),
+                            remote_public.inner(),
+                            private_ips.ipv4.into(),
+                            nonce,
+                        ),
                         wg_port: registration_data.wg_port,
                     },
                     request_id,
@@ -371,7 +378,12 @@ impl<S: Storage + Clone + 'static> MixnetListener<S> {
                 v3::response::AuthenticatorResponse::new_pending_registration_success(
                     v3::registration::RegistrationData {
                         nonce: registration_data.nonce,
-                        gateway_data: registration_data.gateway_data.into(),
+                        gateway_data: v3::registration::GatewayClient::new(
+                            self.keypair().private_key(),
+                            remote_public.inner(),
+                            private_ips.ipv4.into(),
+                            nonce,
+                        ),
                         wg_port: registration_data.wg_port,
                     },
                     request_id,

tools/internal/testnet-manager/dkg-bypass-contract/src/contract.rs

@@ -33,7 +33,7 @@ pub(crate) struct VkShareIndex<'a> {
     pub(crate) epoch_id: MultiIndex<'a, EpochId, ContractVKShare, VKShareKey<'a>>,
 }
 
-impl<'a> IndexList<ContractVKShare> for VkShareIndex<'a> {
+impl IndexList<ContractVKShare> for VkShareIndex<'_> {
     fn get_indexes(&'_ self) -> Box<dyn Iterator<Item = &'_ dyn Index<ContractVKShare>> + '_> {
         let v: Vec<&dyn Index<ContractVKShare>> = vec![&self.epoch_id];
         Box::new(v.into_iter())
@@ -87,7 +87,7 @@ pub fn query(_: Deps<'_>, _: Env, _: EmptyMessage) -> Result<QueryResponse, StdE
 #[cfg_attr(not(feature = "library"), cosmwasm_std::entry_point)]
 pub fn migrate(deps: DepsMut<'_>, env: Env, msg: MigrateMsg) -> Result<Response, StdError> {
     // on migration immediately attempt to rewrite the storage
-    let threshold = (2 * msg.dealers.len() as u64 + 3 - 1) / 3;
+    let threshold = (2 * msg.dealers.len() as u64).div_ceil(3);
     let epoch = CURRENT_EPOCH.load(deps.storage)?;
     assert_eq!(0, epoch.epoch_id);
 

tools/internal/testnet-manager/src/manager/dkg_skip.rs

@@ -58,7 +58,7 @@ impl<'a> FakeDkgKey<'a> {
     }
 }
 
-impl<'a> PemStorableKey for FakeDkgKey<'a> {
+impl PemStorableKey for FakeDkgKey<'_> {
     type Error = NetworkManagerError;
 
     fn pem_type() -> &'static str {
@@ -84,7 +84,7 @@ struct DkgSkipCtx<'a> {
     ecash_signers: Vec<EcashSignerWithPaths>,
 }
 
-impl<'a> ProgressCtx for DkgSkipCtx<'a> {
+impl ProgressCtx for DkgSkipCtx<'_> {
     fn progress_tracker(&self) -> &ProgressTracker {
         &self.progress
     }
@@ -138,7 +138,7 @@ impl NetworkManager {
 
         // generate required materials
         let n = api_endpoints.len();
-        let threshold = (2 * n + 3 - 1) / 3;
+        let threshold = (2 * n).div_ceil(3);
 
         let ecash_keys = ttp_keygen(threshold as u64, n as u64)?;
 

tools/internal/testnet-manager/src/manager/local_apis.rs

@@ -24,7 +24,7 @@ struct LocalApisCtx<'a> {
     signers: Vec<EcashSignerWithPaths>,
 }
 
-impl<'a> ProgressCtx for LocalApisCtx<'a> {
+impl ProgressCtx for LocalApisCtx<'_> {
     fn progress_tracker(&self) -> &ProgressTracker {
         &self.progress
     }

tools/internal/testnet-manager/src/manager/local_client.rs

@@ -30,7 +30,7 @@ struct LocalClientCtx<'a> {
     network: &'a LoadedNetwork,
 }
 
-impl<'a> ProgressCtx for LocalClientCtx<'a> {
+impl ProgressCtx for LocalClientCtx<'_> {
     fn progress_tracker(&self) -> &ProgressTracker {
         &self.progress
     }

tools/internal/testnet-manager/src/manager/local_nodes.rs

@@ -30,7 +30,7 @@ struct LocalNodesCtx<'a> {
     gateways: Vec<NymNode>,
 }
 
-impl<'a> ProgressCtx for LocalNodesCtx<'a> {
+impl ProgressCtx for LocalNodesCtx<'_> {
     fn progress_tracker(&self) -> &ProgressTracker {
         &self.progress
     }