wip

Author: carlosvdr

tap-agent/src/agent/sender_account.rs

@@ -28,7 +28,7 @@ use tracing::{error, Level};
 use super::sender_allocation::{SenderAllocation, SenderAllocationArgs};
 use crate::agent::sender_allocation::SenderAllocationMessage;
 use crate::agent::unaggregated_receipts::UnaggregatedReceipts;
-use crate::tracker::{DurationInfo, SenderFeeStats, SimpleFeeTracker};
+use crate::tracker::{DurationInfo, GlobalFeeTracker, SenderFeeStats, SimpleFeeTracker};
 use crate::{
     config::{self},
     tap::escrow_adapter::EscrowAdapter,
@@ -87,7 +87,7 @@ pub enum ReceiptFees {
     Retry,
 }
 
-type SenderFeeTracker = SimpleFeeTracker<SenderFeeStats, DurationInfo>;
+type SenderFeeTracker = SimpleFeeTracker<SenderFeeStats, DurationInfo, GlobalFeeTracker>;
 
 #[derive(Debug)]
 pub enum SenderAccountMessage {

tap-agent/src/tracker.rs

@@ -11,7 +11,9 @@ mod sender_fee_stats;
 #[cfg(test)]
 mod tracker_tests;
 
-pub type SimpleFeeTracker<T, E = NoExtraData> = GenericTracker<T, E>;
+pub use generic_tracker::GlobalFeeTracker;
+
+pub type SimpleFeeTracker<T, E = NoExtraData, G = u128> = GenericTracker<T, E, G>;
 
 pub trait AllocationStats<G: GlobalTracker> {
     fn update(&mut self, v: G);

tap-agent/src/tracker/generic_tracker.rs

@@ -2,7 +2,7 @@ use super::{AllocationStats, DefaultFromExtra, DurationInfo, SenderFeeStats};
 use alloy::primitives::Address;
 use std::{
     collections::{HashMap, HashSet},
-    ops::AddAssign,
+    ops::{AddAssign, SubAssign},
     time::{Duration, Instant},
 };
 
@@ -16,24 +16,30 @@ impl GlobalTracker for u128 {
     }
 }
 
-type GlobalFeeTracker = u128;
+pub struct GlobalFeeTracker {
+    count: u128,
+    requesting: u128,
+    total_fee: u128,
+}
 
 #[derive(Debug, Clone, Default)]
-pub struct GenericTracker<F, E> {
+pub struct GenericTracker<F, E, G> {
     pub(super) id_to_fee: HashMap<Address, F>,
-    pub(super) total_fee: u128,
+    pub(super) total_fee: G,
     pub(super) extra_data: E,
+    pub(super) fees_requesting: u128,
 }
 
-impl<F, E> GenericTracker<F, E>
+impl<F, E, G> GenericTracker<F, E, G>
 where
-    F: AddAssign<GlobalFeeTracker> + AllocationStats<GlobalFeeTracker> + DefaultFromExtra<E>,
+    G: GlobalTracker + SubAssign + AddAssign + PartialOrd + Copy,
+    F: AddAssign<G> + AllocationStats<G> + DefaultFromExtra<E>,
 {
     /// Updates and overwrite the fee counter into the specific
     /// value provided.
     ///
     /// IMPORTANT: This function does not affect the buffer window fee
-    pub fn update(&mut self, id: Address, value: u128) {
+    pub fn update(&mut self, id: Address, value: G) {
         if !value.should_remove() {
             // insert or update, if update remove old fee from total
             let fee = self
@@ -53,7 +59,7 @@ where
         self.id_to_fee
             .iter_mut()
             .filter(|(_, fee)| !fee.should_filter_out())
-            .fold(None, |acc: Option<(&Address, u128)>, (addr, value)| {
+            .fold(None, |acc: Option<(&Address, G)>, (addr, value)| {
                 if let Some((_, max_fee)) = acc {
                     if value.get_valid_fee() > max_fee {
                         Some((addr, value.get_valid_fee()))
@@ -64,25 +70,29 @@ where
                     Some((addr, value.get_valid_fee()))
                 }
             })
-            .filter(|(_, fee)| *fee > 0)
+            .filter(|(_, fee)| !fee.should_remove())
             .map(|(&id, _)| id)
     }
 
     pub fn get_list_of_allocation_ids(&self) -> HashSet<Address> {
         self.id_to_fee.keys().cloned().collect()
     }
 
-    pub fn get_total_fee(&self) -> u128 {
+    pub fn get_total_fee(&self) -> G {
         self.total_fee
     }
 }
 
-impl GenericTracker<SenderFeeStats, DurationInfo> {
+impl<G> GenericTracker<SenderFeeStats, DurationInfo, G>
+where
+    G: Default,
+{
     pub fn new(buffer_duration: Duration) -> Self {
         Self {
             extra_data: DurationInfo { buffer_duration },
-            total_fee: 0,
+            total_fee: Default::default(),
             id_to_fee: Default::default(),
+            fees_requesting: 0,
         }
     }
 
@@ -91,7 +101,7 @@ impl GenericTracker<SenderFeeStats, DurationInfo> {
     /// It's important to notice that `value` cannot be less than
     /// zero, so the only way to make this counter lower is by using
     /// `update` function
-    pub fn add(&mut self, id: Address, value: u128) {
+    pub fn add(&mut self, id: Address, value: G) {
         let entry = self
             .id_to_fee
             .entry(id)
@@ -114,9 +124,56 @@ impl GenericTracker<SenderFeeStats, DurationInfo> {
             .values_mut()
             .fold(0u128, |acc, expiring| acc + expiring.get_sum())
     }
+
+    pub fn get_total_counter_outside_buffer_for_allocation(
+        &mut self,
+        allocation_id: &Address,
+    ) -> u64 {
+        let Some(allocation_info) = self.id_to_fee.get_mut(allocation_id) else {
+            return 0;
+        };
+        let allocation_counter = allocation_info.count;
+        let counter_in_buffer = allocation_info.get_count();
+        allocation_counter - counter_in_buffer
+    }
+
+    pub fn start_rav_request(&mut self, allocation_id: Address) {
+        let entry = self
+            .id_to_fee
+            .entry(allocation_id)
+            .or_insert(SenderFeeStats::default_from_extra(&self.extra_data));
+        entry.requesting = true;
+        self.fees_requesting += entry.total_fee;
+    }
+
+    /// Should be called before `update`
+    pub fn finish_rav_request(&mut self, allocation_id: Address) {
+        let entry = self
+            .id_to_fee
+            .entry(allocation_id)
+            .or_insert(SenderFeeStats::default_from_extra(&self.extra_data));
+        entry.requesting = false;
+        self.fees_requesting -= entry.total_fee;
+    }
+
+    pub fn failed_rav_backoff(&mut self, allocation_id: Address) {
+        let entry = self
+            .id_to_fee
+            .entry(allocation_id)
+            .or_insert(SenderFeeStats::default_from_extra(&self.extra_data));
+        entry.failed_info.fail();
+    }
+
+    pub fn ok_rav_request(&mut self, allocation_id: Address) {
+        let entry = self
+            .id_to_fee
+            .entry(allocation_id)
+            .or_insert(SenderFeeStats::default_from_extra(&self.extra_data));
+        entry.failed_info.ok();
+    }
 }
 
-impl<E> GenericTracker<SenderFeeStats, E> {
+impl<E, G> GenericTracker<SenderFeeStats, E, G> {
     pub fn block_allocation_id(&mut self, address: Address) {
         self.id_to_fee.entry(address).and_modify(|v| {
             v.blocked = true;
@@ -128,9 +185,16 @@ impl<E> GenericTracker<SenderFeeStats, E> {
             v.blocked = false;
         });
     }
+
+    pub fn check_allocation_has_rav_request_running(&self, allocation_id: Address) -> bool {
+        self.id_to_fee
+            .get(&allocation_id)
+            .map(|alloc| alloc.requesting)
+            .unwrap_or_default()
+    }
 }
 
-impl AllocationStats<GlobalFeeTracker> for u128 {
+impl AllocationStats<u128> for u128 {
     fn update(&mut self, v: u128) {
         *self = v;
     }

tap-agent/src/tracker/sender_fee_stats.rs

@@ -10,15 +10,38 @@ use super::{AllocationStats, DefaultFromExtra, DurationInfo};
 #[derive(Debug, Clone, Default)]
 pub struct SenderFeeStats {
     pub(super) total_fee: u128,
+    pub(super) count: u64,
     // there are some allocations that we don't want it to be
     // heaviest allocation, because they are already marked for finalization,
     // and thus requesting RAVs on their own in their `post_stop` routine.
     pub(super) blocked: bool,
 
+    pub(super) requesting: bool,
+
     // buffer
     pub(super) entries: VecDeque<(Instant, u128)>,
     pub(super) fee_in_buffer: u128,
     pub(super) duration: Duration,
+    pub(super) failed_info: FailedRavInfo,
+}
+#[derive(Debug, Clone)]
+pub struct FailedRavInfo {
+    failed_ravs_count: u32,
+    failed_rav_backoff_time: Instant,
+}
+
+impl FailedRavInfo {
+    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;
+    }
 }
 
 impl DefaultFromExtra<DurationInfo> for SenderFeeStats {
@@ -30,6 +53,15 @@ impl DefaultFromExtra<DurationInfo> for SenderFeeStats {
     }
 }
 
+impl Default for FailedRavInfo {
+    fn default() -> Self {
+        Self {
+            failed_ravs_count: 0,
+            failed_rav_backoff_time: Instant::now(),
+        }
+    }
+}
+
 impl SenderFeeStats {
     pub(super) fn get_sum(&mut self) -> u128 {
         let now = Instant::now();
@@ -43,15 +75,35 @@ impl SenderFeeStats {
         }
         self.fee_in_buffer
     }
+
+    pub(super) fn get_count(&mut self) -> u64 {
+        self.cleanup();
+        self.entries.len() as u64
+    }
+
+    pub(super) fn cleanup(&mut self) {
+        let now = Instant::now();
+        while let Some(&(timestamp, value)) = self.entries.front() {
+            if now.duration_since(timestamp) >= self.duration {
+                self.entries.pop_front();
+                self.fee_in_buffer -= value;
+            } else {
+                break;
+            }
+        }
+    }
 }
 
-impl AllocationStats<u128> for SenderFeeStats {
+impl AllocationStats<GlobalFeeTracker> for SenderFeeStats {
     fn update(&mut self, v: u128) {
         self.total_fee = v;
     }
 
     fn should_filter_out(&self) -> bool {
-        self.blocked
+        let now = Instant::now();
+        let in_backoff = now > self.failed_info.failed_rav_backoff_time;
+
+        in_backoff || self.blocked || self.requesting
     }
 
     fn get_fee(&self) -> u128 {