Add SURBs soft threshold (#5535)

* Add surbs soft threshold

* wip

* Proactively request more SURBs than needed

* fmt

* cleanup

* wip logging

* wip

* debugging

* wip

* Tidy

* tidy

* Set threshold buffer default for IPR

* rustfmt

* wasm fixes

* debug

* Tweak debug message

* Set default min buffer to 0

* Tweak backlog message

* Restore debug message

* tweak

* tweak

* wasm

common/client-core/config-types/src/lib.rs

@@ -45,6 +45,7 @@ const DEFAULT_COVER_TRAFFIC_PRIMARY_SIZE_RATIO: f64 = 0.70;
 // clients/client-core/src/client/replies/reply_storage/surb_storage.rs
 const DEFAULT_MINIMUM_REPLY_SURB_STORAGE_THRESHOLD: usize = 10;
 const DEFAULT_MAXIMUM_REPLY_SURB_STORAGE_THRESHOLD: usize = 200;
+const DEFAULT_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER: usize = 0;
 
 // define how much to request at once
 // clients/client-core/src/client/replies/reply_controller.rs
@@ -621,6 +622,10 @@ pub struct ReplySurbs {
     /// Defines the maximum number of reply surbs the client wants to keep in its storage at any times.
     pub maximum_reply_surb_storage_threshold: usize,
 
+    /// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client
+    /// should proactively request additional reply surbs.
+    pub minimum_reply_surb_threshold_buffer: usize,
+
     /// Defines the minimum number of reply surbs the client would request.
     pub minimum_reply_surb_request_size: u32,
 
@@ -660,6 +665,7 @@ impl Default for ReplySurbs {
         ReplySurbs {
             minimum_reply_surb_storage_threshold: DEFAULT_MINIMUM_REPLY_SURB_STORAGE_THRESHOLD,
             maximum_reply_surb_storage_threshold: DEFAULT_MAXIMUM_REPLY_SURB_STORAGE_THRESHOLD,
+            minimum_reply_surb_threshold_buffer: DEFAULT_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER,
             minimum_reply_surb_request_size: DEFAULT_MINIMUM_REPLY_SURB_REQUEST_SIZE,
             maximum_reply_surb_request_size: DEFAULT_MAXIMUM_REPLY_SURB_REQUEST_SIZE,
             maximum_allowed_reply_surb_request_size: DEFAULT_MAXIMUM_ALLOWED_SURB_REQUEST_SIZE,

common/client-core/config-types/src/old/v5.rs

@@ -181,6 +181,7 @@ impl From<ConfigV5> for Config {
                     maximum_reply_surb_age: value.debug.reply_surbs.maximum_reply_surb_age,
                     maximum_reply_key_age: value.debug.reply_surbs.maximum_reply_key_age,
                     surb_mix_hops: value.debug.reply_surbs.surb_mix_hops,
+                    ..Default::default()
                 },
                 ..Default::default()
             },

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

@@ -517,17 +517,25 @@ where
         use crate::error::ClientCoreStatusMessage;
 
         let packets = self.transmission_buffer.total_size();
-        let backlog = self.transmission_buffer.total_size_in_bytes() as f64 / 1024.0;
-        let lanes = self.transmission_buffer.num_lanes();
+        let lanes = self.transmission_buffer.lanes();
         let mult = self.sending_delay_controller.current_multiplier();
         let delay = self.current_average_message_sending_delay().as_millis();
+
+        let lane_status = lanes
+            .iter()
+            .map(|lane_name| {
+                let lane_length = self.transmission_buffer.lane_length(lane_name).unwrap_or(0);
+                format!("{lane_name:?}: {lane_length}")
+            })
+            .collect::<Vec<String>>()
+            .join(", ");
+
         let status_str = if self.config.traffic.disable_main_poisson_packet_distribution {
-            format!("Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, no delay")
+            format!("Packet backlog: {lane_status}, no delay")
         } else {
-            format!(
-                "Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, avg delay: {delay}ms ({mult})"
-            )
+            format!("Packet backlog: {lane_status}, avg delay: {delay}ms ({mult})")
         };
+
         if packets > 1000 {
             log::warn!("{status_str}");
         } else if packets > 0 {

common/client-core/src/client/replies/reply_controller/mod.rs

@@ -102,6 +102,7 @@ where
         fragments: I,
         lane: TransmissionLane,
     ) {
+        trace!("buffering pending replies for {recipient}");
         self.pending_replies
             .entry(*recipient)
             .or_insert_with(TransmissionBuffer::new)
@@ -113,6 +114,7 @@ where
         recipient: &AnonymousSenderTag,
         fragments: Vec<(TransmissionLane, Fragment)>,
     ) {
+        trace!("re-inserting pending replies for {recipient}");
         // the buffer should ALWAYS exist at this point, if it doesn't, it's a bug...
         self.pending_replies
             .entry(*recipient)
@@ -125,6 +127,7 @@ where
         recipient: &AnonymousSenderTag,
         data: Vec<Arc<PendingAcknowledgement>>,
     ) {
+        trace!("re-inserting pending retransmissions for {recipient}");
         // the underlying entry MUST exist as we've just got data from there
         let map_entry = self
             .pending_retransmissions
@@ -142,7 +145,7 @@ where
     }
 
     fn should_request_more_surbs(&self, target: &AnonymousSenderTag) -> bool {
-        trace!("checking if we should request more surbs from {:?}", target);
+        trace!("checking if we should request more surbs from {target}");
 
         let pending_queue_size = self
             .pending_replies
@@ -158,11 +161,6 @@ where
 
         let total_queue = pending_queue_size + retransmission_queue;
 
-        // simple as that - there's absolutely nothing to retransmit
-        if total_queue == 0 {
-            return false;
-        }
-
         let available_surbs = self
             .full_reply_storage
             .surbs_storage_ref()
@@ -179,11 +177,27 @@ where
             .full_reply_storage
             .surbs_storage_ref()
             .max_surb_threshold();
+        let min_surbs_threshold_buffer =
+            self.config.reply_surbs.minimum_reply_surb_threshold_buffer;
 
-        debug!("total queue size: {total_queue} = pending data {pending_queue_size} + pending retransmission {retransmission_queue}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..{max_surbs_threshold}");
+        // After clearing the queue, we want to have at least `min_surbs_threshold` surbs available
+        // and reserved for requesting additional surbs, and in addition to that we also want to
+        // have `min_surbs_threshold_buffer` surbs available proactively.
+        let target_surbs_after_clearing_queue = min_surbs_threshold + min_surbs_threshold_buffer;
 
-        (pending_surbs + available_surbs) < max_surbs_threshold
-            && (pending_surbs + available_surbs) < (total_queue + min_surbs_threshold)
+        // Check if we have enough surbs to handle the total queue and maintain minimum thresholds
+        let total_required_surbs = total_queue + target_surbs_after_clearing_queue;
+        let total_available_surbs = pending_surbs + available_surbs;
+
+        debug!("total queue size: {total_queue} = pending data {pending_queue_size} + pending retransmission {retransmission_queue}, available surbs: {available_surbs} pending surbs: {pending_surbs} threshold range: {min_surbs_threshold}..+{min_surbs_threshold_buffer}..{max_surbs_threshold}");
+
+        // We should request more surbs if:
+        // 1. We haven't hit the maximum surb threshold, and
+        // 2. We don't have enough surbs to handle the queue plus minimum thresholds
+        let is_below_max_threshold = total_available_surbs < max_surbs_threshold;
+        let is_below_required_surbs = total_available_surbs < total_required_surbs;
+
+        is_below_max_threshold && is_below_required_surbs
     }
 
     async fn handle_send_reply(
@@ -244,6 +258,10 @@ where
                         &recipient_tag,
                     );
                     warn!("failed to send reply to {recipient_tag}: {err}");
+                    info!(
+                        "buffering {no_fragments} fragments for {recipient_tag}",
+                        no_fragments = to_send.len()
+                    );
                     self.insert_pending_replies(&recipient_tag, to_send, lane);
                 }
             }
@@ -251,6 +269,13 @@ where
 
         // if there's leftover data we didn't send because we didn't have enough (or any) surbs - buffer it
         if !fragments.is_empty() {
+            // Ideally we should have enough surbs above the minimum threshold to handle sending
+            // new replies without having to first request more surbs. That's why I'd like to log
+            // these cases as they might indicate a problem with the surb management.
+            debug!(
+                "buffering {no_fragments} fragments for {recipient_tag}",
+                no_fragments = fragments.len()
+            );
             self.insert_pending_replies(&recipient_tag, fragments, lane);
         }
 
@@ -265,6 +290,7 @@ where
         target: AnonymousSenderTag,
         amount: u32,
     ) -> Result<(), PreparationError> {
+        debug!("requesting {amount} additional reply surbs for {target}");
         let reply_surb = self
             .full_reply_storage
             .surbs_storage_ref()
@@ -686,7 +712,7 @@ where
     // it should take into consideration the average latency, sending rate and queue size.
     // it should request as many surbs as it takes to saturate its sending rate before next batch arrives
     async fn request_reply_surbs_for_queue_clearing(&mut self, target: AnonymousSenderTag) {
-        trace!("requesting surbs for queues clearing");
+        trace!("requesting surbs for queue clearing");
 
         let pending_queue_size = self
             .pending_replies
@@ -700,17 +726,18 @@ where
             .map(|pending_queue| pending_queue.len())
             .unwrap_or_default();
 
+        let min_surbs_buffer = self.config.reply_surbs.minimum_reply_surb_threshold_buffer as u32;
+
         let total_queue = (pending_queue_size + retransmission_queue) as u32;
 
-        if total_queue == 0 {
-            trace!("the pending queues for {:?} are already empty", target);
-            return;
-        }
+        // To proactively request additional surbs, we aim to have a buffer of extra surbs in our
+        // storage.
+        let total_queue_with_buffer = total_queue + min_surbs_buffer;
 
         let request_size = min(
             self.config.reply_surbs.maximum_reply_surb_request_size,
             max(
-                total_queue,
+                total_queue_with_buffer,
                 self.config.reply_surbs.minimum_reply_surb_request_size,
             ),
         );

common/client-core/src/client/transmission_buffer.rs

@@ -58,8 +58,8 @@ impl<T> TransmissionBuffer<T> {
     }
 
     #[cfg(not(target_arch = "wasm32"))]
-    pub(crate) fn num_lanes(&self) -> usize {
-        self.buffer.keys().count()
+    pub(crate) fn lanes(&self) -> Vec<TransmissionLane> {
+        self.buffer.keys().cloned().collect()
     }
 
     pub(crate) fn lane_length(&self, lane: &TransmissionLane) -> Option<usize> {
@@ -83,6 +83,7 @@ impl<T> TransmissionBuffer<T> {
     }
 
     #[cfg(not(target_arch = "wasm32"))]
+    #[allow(unused)]
     pub(crate) fn total_size_in_bytes(&self) -> usize
     where
         T: SizedData,

common/wasm/client-core/src/config/mod.rs

@@ -462,6 +462,10 @@ pub struct ReplySurbsWasm {
     /// Defines the maximum number of reply surbs the client wants to keep in its storage at any times.
     pub maximum_reply_surb_storage_threshold: usize,
 
+    /// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client
+    /// should proactively request additional reply surbs.
+    pub minimum_reply_surb_threshold_buffer: usize,
+
     /// Defines the minimum number of reply surbs the client would request.
     pub minimum_reply_surb_request_size: u32,
 
@@ -503,6 +507,7 @@ impl From<ReplySurbsWasm> for ConfigReplySurbs {
         ConfigReplySurbs {
             minimum_reply_surb_storage_threshold: reply_surbs.minimum_reply_surb_storage_threshold,
             maximum_reply_surb_storage_threshold: reply_surbs.maximum_reply_surb_storage_threshold,
+            minimum_reply_surb_threshold_buffer: reply_surbs.minimum_reply_surb_threshold_buffer,
             minimum_reply_surb_request_size: reply_surbs.minimum_reply_surb_request_size,
             maximum_reply_surb_request_size: reply_surbs.maximum_reply_surb_request_size,
             maximum_allowed_reply_surb_request_size: reply_surbs
@@ -529,6 +534,7 @@ impl From<ConfigReplySurbs> for ReplySurbsWasm {
         ReplySurbsWasm {
             minimum_reply_surb_storage_threshold: reply_surbs.minimum_reply_surb_storage_threshold,
             maximum_reply_surb_storage_threshold: reply_surbs.maximum_reply_surb_storage_threshold,
+            minimum_reply_surb_threshold_buffer: reply_surbs.minimum_reply_surb_threshold_buffer,
             minimum_reply_surb_request_size: reply_surbs.minimum_reply_surb_request_size,
             maximum_reply_surb_request_size: reply_surbs.maximum_reply_surb_request_size,
             maximum_allowed_reply_surb_request_size: reply_surbs

common/wasm/client-core/src/config/override.rs

@@ -339,6 +339,11 @@ pub struct ReplySurbsWasmOverride {
     #[tsify(optional)]
     pub maximum_reply_surb_storage_threshold: Option<usize>,
 
+    /// Defines the soft threshold ontop of the minimum reply surb storage threshold for when the client
+    /// should proactively request additional reply surbs.
+    #[tsify(optional)]
+    pub minimum_reply_surb_threshold_buffer: Option<usize>,
+
     /// Defines the minimum number of reply surbs the client would request.
     #[tsify(optional)]
     pub minimum_reply_surb_request_size: Option<u32>,
@@ -386,6 +391,9 @@ impl From<ReplySurbsWasmOverride> for ReplySurbsWasm {
             maximum_reply_surb_storage_threshold: value
                 .maximum_reply_surb_storage_threshold
                 .unwrap_or(def.maximum_reply_surb_storage_threshold),
+            minimum_reply_surb_threshold_buffer: value
+                .minimum_reply_surb_threshold_buffer
+                .unwrap_or(def.minimum_reply_surb_threshold_buffer),
             minimum_reply_surb_request_size: value
                 .minimum_reply_surb_request_size
                 .unwrap_or(def.minimum_reply_surb_request_size),

nym-node/src/config/service_providers.rs

@@ -3,12 +3,17 @@
 
 use crate::config::authenticator::Authenticator;
 use crate::config::persistence::ServiceProvidersPaths;
-use nym_client_core_config_types::DebugConfig as ClientDebugConfig;
+use nym_client_core_config_types::{DebugConfig as ClientDebugConfig, ReplySurbs};
 use nym_config::defaults::mainnet;
 use serde::{Deserialize, Serialize};
 use std::path::Path;
 use url::Url;
 
+// Default reply surb storage threshold.
+// It is set higher than the default since the traffic is mostly a stream of data,
+// and not a bursty request/response pattern.
+pub const DEFAULT_IPR_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER: usize = 50;
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ServiceProvidersConfig {
@@ -145,10 +150,18 @@ pub struct IpPacketRouterDebug {
 
 impl Default for IpPacketRouterDebug {
     fn default() -> Self {
+        let client_debug = ClientDebugConfig {
+            reply_surbs: ReplySurbs {
+                minimum_reply_surb_threshold_buffer:
+                    DEFAULT_IPR_MINIMUM_REPLY_SURB_THRESHOLD_BUFFER,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
         IpPacketRouterDebug {
             enabled: true,
             disable_poisson_rate: true,
-            client_debug: Default::default(),
+            client_debug,
         }
     }
 }