Make RAV value threshold per sender (#113)

* Make RAV value threshold per sender Fixes #93 Signed-off-by: Alexis Asseman <[email protected]> * fix: simplify rav_requester_finalize error handling Signed-off-by: Alexis Asseman <[email protected]> * fix: proper signer/sender handling at startup DB SenderAccount loading Signed-off-by: Alexis Asseman <[email protected]> * test: cleanup Signed-off-by: Alexis Asseman <[email protected]> * fix: nested SQL query in sender_accounts_manager Signed-off-by: Alexis Asseman <[email protected]> * refactor: move static methods into Inner impl Signed-off-by: Alexis Asseman <[email protected]> * refactor: simplify SenderAcccount's fees_add Signed-off-by: Alexis Asseman <[email protected]> * refactor: replace active and ineligible allocations with AllocationStateEnum and single HashMap Signed-off-by: Alexis Asseman <[email protected]> * fix: locking issue and recompute unaggregated fees * fix: unaggregated fees update race condition with tokio mutex guards Signed-off-by: Alexis Asseman <[email protected]> * chore: fix comments Signed-off-by: Alexis Asseman <[email protected]> * fix: use std mutex for sender_accounts Signed-off-by: Alexis Asseman <[email protected]> --------- Signed-off-by: Alexis Asseman <[email protected]> Co-authored-by: Gustavo Inacio <[email protected]>
graphprotocol · Feb 16, 2024 · 308562e · 308562e
1 parent d3b58eb
commit 308562e
Show file tree

Hide file tree

Showing 15 changed files with 1,893 additions and 1,243 deletions.
diff --git a/.sqlx/query-08726aaece34702a956885e927166df6b1f8ad0a0589464b9cbd9f09ca3f38db.json b/.sqlx/query-08726aaece34702a956885e927166df6b1f8ad0a0589464b9cbd9f09ca3f38db.json
diff --git a/.sqlx/query-778a427621acd2003b94340e46df1b73bfc863f0fa48004a4d5bd39cd97b07bb.json b/.sqlx/query-778a427621acd2003b94340e46df1b73bfc863f0fa48004a4d5bd39cd97b07bb.json
diff --git a/.sqlx/query-9fb8ca8ec1553951f20f81acdcfe52ad56d1bbac73fbcf5a93808bd4eaafdd4a.json b/.sqlx/query-9fb8ca8ec1553951f20f81acdcfe52ad56d1bbac73fbcf5a93808bd4eaafdd4a.json
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/tap-agent/Cargo.toml b/tap-agent/Cargo.toml
@@ -46,6 +46,7 @@ tracing-subscriber = { version = "0.3", features = [
     "std",
     "json",
 ] }
+enum-as-inner = "0.6.0"
 
 [dev-dependencies]
 ethers-signers = "2.0.8"

diff --git a/tap-agent/src/agent.rs b/tap-agent/src/agent.rs
@@ -9,11 +9,10 @@ use indexer_common::prelude::{
 };
 
 use crate::{
-    aggregator_endpoints, config, database,
-    tap::sender_allocation_relationships_manager::SenderAllocationRelationshipsManager,
+    aggregator_endpoints, config, database, tap::sender_accounts_manager::SenderAccountsManager,
 };
 
-pub async fn start_agent(config: &'static config::Cli) -> SenderAllocationRelationshipsManager {
+pub async fn start_agent(config: &'static config::Cli) -> SenderAccountsManager {
     let pgpool = database::connect(&config.postgres).await;
 
     let http_client = reqwest::Client::new();
@@ -80,7 +79,7 @@ pub async fn start_agent(config: &'static config::Cli) -> SenderAllocationRelati
         verifying_contract: config.receipts.receipts_verifier_address,
     };
 
-    SenderAllocationRelationshipsManager::new(
+    SenderAccountsManager::new(
         config,
         pgpool,
         indexer_allocations,

diff --git a/tap-agent/src/tap/mod.rs b/tap-agent/src/tap/mod.rs
@@ -11,8 +11,10 @@ mod escrow_adapter;
 mod rav_storage_adapter;
 mod receipt_checks_adapter;
 mod receipt_storage_adapter;
-mod sender_allocation_relationship;
-pub mod sender_allocation_relationships_manager;
+mod sender_account;
+pub mod sender_accounts_manager;
+mod sender_allocation;
+mod unaggregated_receipts;
 
 #[cfg(test)]
 pub mod test_utils;

diff --git a/tap-agent/src/tap/rav_storage_adapter.rs b/tap-agent/src/tap/rav_storage_adapter.rs
@@ -91,18 +91,18 @@ impl RAVStorageAdapter {
 #[cfg(test)]
 mod test {
     use super::*;
-    use crate::tap::test_utils::{create_rav, ALLOCATION_ID, SENDER, SIGNER};
+    use crate::tap::test_utils::{create_rav, ALLOCATION_ID_0, SENDER, SIGNER};
     use tap_core::adapters::rav_storage_adapter::RAVStorageAdapter as RAVStorageAdapterTrait;
 
     #[sqlx::test(migrations = "../migrations")]
     async fn update_and_retrieve_rav(pool: PgPool) {
         let timestamp_ns = u64::MAX - 10;
         let value_aggregate = u128::MAX;
-        let rav_storage_adapter = RAVStorageAdapter::new(pool.clone(), *ALLOCATION_ID, SENDER.1);
+        let rav_storage_adapter = RAVStorageAdapter::new(pool.clone(), *ALLOCATION_ID_0, SENDER.1);
 
         // Insert a rav
         let mut new_rav = create_rav(
-            *ALLOCATION_ID,
+            *ALLOCATION_ID_0,
             SIGNER.0.clone(),
             timestamp_ns,
             value_aggregate,
@@ -121,7 +121,7 @@ mod test {
         // Update the RAV 3 times in quick succession
         for i in 0..3 {
             new_rav = create_rav(
-                *ALLOCATION_ID,
+                *ALLOCATION_ID_0,
                 SIGNER.0.clone(),
                 timestamp_ns + i,
                 value_aggregate - (i as u128),

diff --git a/tap-agent/src/tap/receipt_storage_adapter.rs b/tap-agent/src/tap/receipt_storage_adapter.rs
@@ -194,7 +194,7 @@ mod test {
 
     use super::*;
     use crate::tap::test_utils::{
-        create_received_receipt, store_receipt, ALLOCATION_ID, ALLOCATION_ID_IRRELEVANT, SENDER,
+        create_received_receipt, store_receipt, ALLOCATION_ID_0, ALLOCATION_ID_IRRELEVANT, SENDER,
         SENDER_IRRELEVANT, SIGNER, TAP_EIP712_DOMAIN_SEPARATOR,
     };
     use anyhow::Result;
@@ -381,7 +381,7 @@ mod test {
 
         let storage_adapter = ReceiptStorageAdapter::new(
             pgpool.clone(),
-            *ALLOCATION_ID,
+            *ALLOCATION_ID_0,
             SENDER.1,
             get_full_list_of_checks(),
             escrow_accounts.clone(),
@@ -392,7 +392,7 @@ mod test {
         for i in 0..10 {
             received_receipt_vec.push(
                 create_received_receipt(
-                    &ALLOCATION_ID,
+                    &ALLOCATION_ID_0,
                     &SIGNER.0,
                     i + 684,
                     i + 42,
@@ -416,7 +416,7 @@ mod test {
             );
             received_receipt_vec.push(
                 create_received_receipt(
-                    &ALLOCATION_ID,
+                    &ALLOCATION_ID_0,
                     &SENDER_IRRELEVANT.0,
                     i + 684,
                     i + 42,
@@ -525,7 +525,7 @@ mod test {
 
         let storage_adapter = ReceiptStorageAdapter::new(
             pgpool,
-            *ALLOCATION_ID,
+            *ALLOCATION_ID_0,
             SENDER.1,
             get_full_list_of_checks(),
             escrow_accounts.clone(),
@@ -536,7 +536,7 @@ mod test {
         for i in 0..10 {
             received_receipt_vec.push(
                 create_received_receipt(
-                    &ALLOCATION_ID,
+                    &ALLOCATION_ID_0,
                     &SIGNER.0,
                     i + 684,
                     i + 42,
@@ -560,7 +560,7 @@ mod test {
             );
             received_receipt_vec.push(
                 create_received_receipt(
-                    &ALLOCATION_ID,
+                    &ALLOCATION_ID_0,
                     &SENDER_IRRELEVANT.0,
                     i + 684,
                     i + 42,
-Original file line number
+Diff line change
@@ Expand Up / @@ -46,6 +46,7 @@ tracing-subscriber = { version = "0.3", features = [ @@
         "std",
         "json",
     ] }
+    enum-as-inner = "0.6.0"
     [dev-dependencies]
     ethers-signers = "2.0.8"
@@ Expand Down @@