fix: add backoff if could not find heaviest allocation (#397)

Author: carlosvdr

tap-agent/src/agent/sender_account.rs

@@ -30,6 +30,7 @@ use super::sender_allocation::{SenderAllocation, SenderAllocationArgs};
 use crate::adaptative_concurrency::AdaptiveLimiter;
 use crate::agent::sender_allocation::SenderAllocationMessage;
 use crate::agent::unaggregated_receipts::UnaggregatedReceipts;
+use crate::backoff::BackoffInfo;
 use crate::tracker::{SenderFeeTracker, SimpleFeeTracker};
 use crate::{
     config::{self},
@@ -161,6 +162,9 @@ pub struct State {
     config: &'static config::Config,
     pgpool: PgPool,
     sender_aggregator: jsonrpsee::http_client::HttpClient,
+
+    // Backoff info
+    backoff_info: BackoffInfo,
 }
 
 impl State {
@@ -211,6 +215,7 @@ impl State {
             .sender_fee_tracker
             .get_heaviest_allocation_id()
             .ok_or_else(|| {
+                self.backoff_info.fail();
                 anyhow::anyhow!(
                     "Error while getting the heaviest allocation, \
             this is due one of the following reasons: \n
@@ -221,6 +226,7 @@ impl State {
             If this doesn't work, open an issue on our Github."
                 )
             })?;
+        self.backoff_info.ok();
         self.rav_request_for_allocation(allocation_id).await
     }
 
@@ -546,6 +552,7 @@ impl Actor for SenderAccount {
             retry_interval,
             scheduled_rav_request: None,
             adaptive_limiter: AdaptiveLimiter::new(INITIAL_RAV_REQUEST_CONCURRENT, 1..50),
+            backoff_info: BackoffInfo::default(),
         };
 
         for allocation_id in &allocation_ids {
@@ -650,39 +657,33 @@ impl Actor for SenderAccount {
 
                 let has_available_slots_for_requests = state.adaptive_limiter.has_limit();
                 if has_available_slots_for_requests {
+                    let total_fee_outside_buffer = state.sender_fee_tracker.get_ravable_total_fee();
                     let total_counter_for_allocation = state
                         .sender_fee_tracker
                         .get_count_outside_buffer_for_allocation(&allocation_id);
                     let can_trigger_rav = state.sender_fee_tracker.can_trigger_rav(allocation_id);
                     let counter_greater_receipt_limit = total_counter_for_allocation
                         >= state.config.tap.rav_request_receipt_limit
                         && can_trigger_rav;
-                    let total_fee_outside_buffer = state.sender_fee_tracker.get_ravable_total_fee();
-                    let total_fee_greater_trigger_value =
-                        total_fee_outside_buffer >= state.config.tap.rav_request_trigger_value;
-                    let rav_result = match (
-                        counter_greater_receipt_limit,
-                        total_fee_greater_trigger_value,
-                    ) {
-                        (true, _) => {
-                            tracing::debug!(
-                                total_counter_for_allocation,
-                                rav_request_receipt_limit = state.config.tap.rav_request_receipt_limit,
-                                %allocation_id,
-                                "Total counter greater than the receipt limit per rav. Triggering RAV request"
-                            );
-
-                            state.rav_request_for_allocation(allocation_id).await
-                        }
-                        (_, true) => {
-                            tracing::debug!(
-                                total_fee_outside_buffer,
-                                trigger_value = state.config.tap.rav_request_trigger_value,
-                                "Total fee greater than the trigger value. Triggering RAV request"
-                            );
-                            state.rav_request_for_heaviest_allocation().await
-                        }
-                        _ => Ok(()),
+                    let rav_result = if !state.backoff_info.in_backoff()
+                        && total_fee_outside_buffer >= state.config.tap.rav_request_trigger_value
+                    {
+                        tracing::debug!(
+                            total_fee_outside_buffer,
+                            trigger_value = state.config.tap.rav_request_trigger_value,
+                            "Total fee greater than the trigger value. Triggering RAV request"
+                        );
+                        state.rav_request_for_heaviest_allocation().await
+                    } else if counter_greater_receipt_limit {
+                        tracing::debug!(
+                            total_counter_for_allocation,
+                            rav_request_receipt_limit = state.config.tap.rav_request_receipt_limit,
+                            %allocation_id,
+                            "Total counter greater than the receipt limit per rav. Triggering RAV request"
+                        );
+                        state.rav_request_for_allocation(allocation_id).await
+                    } else {
+                        Ok(())
                     };
                     // In case we fail, we want our actor to keep running
                     if let Err(err) = rav_result {

tap-agent/src/backoff.rs

@@ -0,0 +1,38 @@
+// Copyright 2023-, Edge & Node, GraphOps, and Semiotic Labs.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::time::{Duration, Instant};
+
+#[derive(Debug, Clone)]
+pub struct BackoffInfo {
+    failed_count: u32,
+    failed_backoff_time: Instant,
+}
+
+impl BackoffInfo {
+    pub fn ok(&mut self) {
+        self.failed_count = 0;
+    }
+
+    pub fn fail(&mut self) {
+        // backoff = max(100ms * 2 ^ retries, 60s)
+        self.failed_backoff_time = Instant::now()
+            + (Duration::from_millis(100) * 2u32.pow(self.failed_count))
+                .min(Duration::from_secs(60));
+        self.failed_count += 1;
+    }
+
+    pub fn in_backoff(&self) -> bool {
+        let now = Instant::now();
+        now < self.failed_backoff_time
+    }
+}
+
+impl Default for BackoffInfo {
+    fn default() -> Self {
+        Self {
+            failed_count: 0,
+            failed_backoff_time: Instant::now(),
+        }
+    }
+}

tap-agent/src/lib.rs

@@ -17,6 +17,7 @@ lazy_static! {
 
 pub mod adaptative_concurrency;
 pub mod agent;
+pub mod backoff;
 pub mod config;
 pub mod database;
 pub mod metrics;

tap-agent/src/tracker/sender_fee_stats.rs

@@ -3,10 +3,10 @@
 
 use std::{
     collections::VecDeque,
-    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+    time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
-use crate::agent::unaggregated_receipts::UnaggregatedReceipts;
+use crate::{agent::unaggregated_receipts::UnaggregatedReceipts, backoff::BackoffInfo};
 
 use super::{AllocationStats, DefaultFromExtra, DurationInfo};
 
@@ -84,31 +84,6 @@ impl BufferInfo {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct BackoffInfo {
-    failed_ravs_count: u32,
-    failed_rav_backoff_time: Instant,
-}
-
-impl BackoffInfo {
-    pub fn ok(&mut self) {
-        self.failed_ravs_count = 0;
-    }
-
-    pub fn fail(&mut self) {
-        // backoff = max(100ms * 2 ^ retries, 60s)
-        self.failed_rav_backoff_time = Instant::now()
-            + (Duration::from_millis(100) * 2u32.pow(self.failed_ravs_count))
-                .min(Duration::from_secs(60));
-        self.failed_ravs_count += 1;
-    }
-
-    pub fn in_backoff(&self) -> bool {
-        let now = Instant::now();
-        now < self.failed_rav_backoff_time
-    }
-}
-
 impl DefaultFromExtra<DurationInfo> for SenderFeeStats {
     fn default_from_extra(extra: &DurationInfo) -> Self {
         SenderFeeStats {
@@ -121,15 +96,6 @@ impl DefaultFromExtra<DurationInfo> for SenderFeeStats {
     }
 }
 
-impl Default for BackoffInfo {
-    fn default() -> Self {
-        Self {
-            failed_ravs_count: 0,
-            failed_rav_backoff_time: Instant::now(),
-        }
-    }
-}
-
 impl AllocationStats<UnaggregatedReceipts> for SenderFeeStats {
     fn update(&mut self, v: UnaggregatedReceipts) {
         self.total_fee = v.value;