refactor: update check escrow receipt check

Signed-off-by: Gustavo Inacio <[email protected]>

Author: gusinacio

Cargo.lock

@@ -76,7 +76,7 @@ graphql_client = { version = "0.14.0", features = ["reqwest-rustls"] }
 bip39 = "2.0.0"
 rstest = "0.23.0"
 wiremock = "0.6.1"
-typed-builder = "0.20.0"
+bon = "3.3"
 tonic = { version = "0.12.3", features = ["tls-roots", "gzip"] }
 prost = "0.13.4"
 prost-types = "0.13.3"

Cargo.toml

@@ -88,7 +88,7 @@ impl EscrowAccounts {
 
 pub type EscrowAccountsWatcher = Receiver<EscrowAccounts>;
 
-pub async fn escrow_accounts(
+pub async fn escrow_accounts_v1(
     escrow_subgraph: &'static SubgraphClient,
     indexer_address: Address,
     interval: Duration,
@@ -243,7 +243,7 @@ mod tests {
             );
         mock_server.register(mock).await;
 
-        let mut accounts = escrow_accounts(
+        let mut accounts = escrow_accounts_v1(
             escrow_subgraph,
             test_assets::INDEXER_ADDRESS,
             Duration::from_secs(60),

crates/monitor/src/escrow_accounts.rs

@@ -15,7 +15,7 @@ pub use crate::{
     deployment_to_allocation::{deployment_to_allocation, DeploymentToAllocationWatcher},
     dispute_manager::{dispute_manager, DisputeManagerWatcher},
     escrow_accounts::{
-        escrow_accounts, escrow_accounts_v2, EscrowAccounts, EscrowAccountsError,
+        escrow_accounts_v1, escrow_accounts_v2, EscrowAccounts, EscrowAccountsError,
         EscrowAccountsWatcher,
     },
 };

crates/monitor/src/lib.rs

@@ -40,7 +40,7 @@ graphql = { git = "https://github.com/edgeandnode/toolshed", tag = "graphql-v0.3
 tap_core.workspace = true
 tap_graph.workspace = true
 uuid.workspace = true
-typed-builder.workspace = true
+bon.workspace = true
 tower_governor = { version = "0.5.0", features = ["axum"] }
 governor = "0.8.0"
 tower-http = { version = "0.6.2", features = [

crates/service/Cargo.toml

@@ -17,8 +17,10 @@ use crate::{error::IndexerServiceError, tap::TapReceipt};
 pub struct SenderState {
     /// Used to recover the signer address
     pub domain_separator: Eip712Domain,
-    /// Used to get the sender address given the signer address
-    pub escrow_accounts: watch::Receiver<EscrowAccounts>,
+    /// Used to get the sender address given the signer address if v1 receipt
+    pub escrow_accounts_v1: watch::Receiver<EscrowAccounts>,
+    /// Used to get the sender address given the signer address if v2 receipt
+    pub escrow_accounts_v2: watch::Receiver<EscrowAccounts>,
 }
 
 /// The current query Sender address
@@ -45,10 +47,16 @@ pub async fn sender_middleware(
 ) -> Result<Response, IndexerServiceError> {
     if let Some(receipt) = request.extensions().get::<TapReceipt>() {
         let signer = receipt.recover_signer(&state.domain_separator)?;
-        let sender = state
-            .escrow_accounts
-            .borrow()
-            .get_sender_for_signer(&signer)?;
+        let sender = match receipt {
+            TapReceipt::V1(_) => state
+                .escrow_accounts_v1
+                .borrow()
+                .get_sender_for_signer(&signer)?,
+            TapReceipt::V2(_) => state
+                .escrow_accounts_v2
+                .borrow()
+                .get_sender_for_signer(&signer)?,
+        };
         request.extensions_mut().insert(Sender(sender));
     }
 
@@ -78,14 +86,22 @@ mod tests {
 
     #[tokio::test]
     async fn test_sender_middleware() {
-        let escrow_accounts = watch::channel(EscrowAccounts::new(
+        let escrow_accounts_v1 = watch::channel(EscrowAccounts::new(
             ESCROW_ACCOUNTS_BALANCES.to_owned(),
             ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.to_owned(),
         ))
         .1;
+
+        let escrow_accounts_v2 = watch::channel(EscrowAccounts::new(
+            ESCROW_ACCOUNTS_BALANCES.to_owned(),
+            ESCROW_ACCOUNTS_SENDERS_TO_SIGNERS.to_owned(),
+        ))
+        .1;
+
         let state = SenderState {
             domain_separator: test_assets::TAP_EIP712_DOMAIN.clone(),
-            escrow_accounts,
+            escrow_accounts_v1,
+            escrow_accounts_v2,
         };
 
         let middleware = from_fn_with_state(state, sender_middleware);

crates/service/src/middleware/sender.rs

@@ -17,9 +17,9 @@ use indexer_config::{
     ServiceConfig, ServiceTapConfig,
 };
 use indexer_monitor::{
-    attestation_signers, deployment_to_allocation, dispute_manager, escrow_accounts,
-    indexer_allocations, AllocationWatcher, DisputeManagerWatcher, EscrowAccountsWatcher,
-    SubgraphClient,
+    attestation_signers, deployment_to_allocation, dispute_manager, escrow_accounts_v1,
+    escrow_accounts_v2, indexer_allocations, AllocationWatcher, DisputeManagerWatcher,
+    EscrowAccountsWatcher, SubgraphClient,
 };
 use reqwest::Method;
 use tap_core::{manager::Manager, receipt::checks::CheckList};
@@ -34,7 +34,6 @@ use tower_http::{
     trace::TraceLayer,
     validate_request::ValidateRequestHeaderLayer,
 };
-use typed_builder::TypedBuilder;
 
 use super::{release::IndexerServiceRelease, GraphNodeState};
 use crate::{
@@ -51,7 +50,7 @@ use crate::{
     wallet::public_key,
 };
 
-#[derive(TypedBuilder)]
+#[derive(bon::Builder)]
 pub struct ServiceRouter {
     // database
     database: sqlx::PgPool,
@@ -60,7 +59,6 @@ pub struct ServiceRouter {
     // graphnode client
     http_client: reqwest::Client,
     // release info
-    #[builder(default, setter(strip_option))]
     release: Option<IndexerServiceRelease>,
 
     // configuration
@@ -71,23 +69,21 @@ pub struct ServiceRouter {
     timestamp_buffer_secs: Duration,
 
     // either provide subgraph or watcher
-    #[builder(default, setter(transform =
+    #[builder(with =
         |subgraph: &'static SubgraphClient,
-            config: EscrowSubgraphConfig|
-        Some((subgraph, config))))]
+        config: EscrowSubgraphConfig|
+        (subgraph, config))]
     escrow_subgraph: Option<(&'static SubgraphClient, EscrowSubgraphConfig)>,
-    #[builder(default, setter(strip_option))]
-    escrow_accounts: Option<EscrowAccountsWatcher>,
+    escrow_accounts_v1: Option<EscrowAccountsWatcher>,
+
+    escrow_accounts_v2: Option<EscrowAccountsWatcher>,
 
     // provide network subgraph or allocations + dispute manager
-    #[builder(default, setter(transform =
-        |subgraph: &'static SubgraphClient,
-            config: NetworkSubgraphConfig|
-        Some((subgraph, config))))]
+    #[builder(with = |subgraph: &'static SubgraphClient,
+        config: NetworkSubgraphConfig|
+        (subgraph, config))]
     network_subgraph: Option<(&'static SubgraphClient, NetworkSubgraphConfig)>,
-    #[builder(default, setter(strip_option))]
     allocations: Option<AllocationWatcher>,
-    #[builder(default, setter(strip_option))]
     dispute_manager: Option<DisputeManagerWatcher>,
 }
 
@@ -141,11 +137,26 @@ impl ServiceRouter {
             (None, None) => panic!("No allocations or network subgraph was provided"),
         };
 
-        // Monitor escrow accounts
+        // Monitor escrow accounts v1
+        // if not provided, create monitor from subgraph
+        let escrow_accounts_v1 = match (self.escrow_accounts_v1, self.escrow_subgraph.as_ref()) {
+            (Some(escrow_account), _) => escrow_account,
+            (_, Some((escrow_subgraph, escrow))) => escrow_accounts_v1(
+                escrow_subgraph,
+                indexer_address,
+                escrow.config.syncing_interval_secs,
+                true, // Reject thawing signers eagerly
+            )
+            .await
+            .expect("Error creating escrow_accounts channel"),
+            (None, None) => panic!("No escrow accounts or escrow subgraph was provided"),
+        };
+
+        // Monitor escrow accounts v2
         // if not provided, create monitor from subgraph
-        let escrow_accounts = match (self.escrow_accounts, self.escrow_subgraph.as_ref()) {
+        let escrow_accounts_v2 = match (self.escrow_accounts_v2, self.escrow_subgraph.as_ref()) {
             (Some(escrow_account), _) => escrow_account,
-            (_, Some((escrow_subgraph, escrow))) => escrow_accounts(
+            (_, Some((escrow_subgraph, escrow))) => escrow_accounts_v2(
                 escrow_subgraph,
                 indexer_address,
                 escrow.config.syncing_interval_secs,
@@ -255,7 +266,8 @@ impl ServiceRouter {
                 let checks = IndexerTapContext::get_checks(
                     self.database,
                     allocations.clone(),
-                    escrow_accounts.clone(),
+                    escrow_accounts_v1.clone(),
+                    escrow_accounts_v2.clone(),
                     timestamp_error_tolerance,
                     receipt_max_value,
                 )
@@ -299,7 +311,8 @@ impl ServiceRouter {
                 deployment_to_allocation,
             };
             let sender_state = SenderState {
-                escrow_accounts,
+                escrow_accounts_v1,
+                escrow_accounts_v2,
                 domain_separator: self.domain_separator,
             };
 

crates/service/src/service/router.rs

@@ -48,13 +48,17 @@ impl IndexerTapContext {
     pub async fn get_checks(
         pgpool: PgPool,
         indexer_allocations: Receiver<HashMap<Address, Allocation>>,
-        escrow_accounts: Receiver<EscrowAccounts>,
+        escrow_accounts_v1: Receiver<EscrowAccounts>,
+        escrow_accounts_v2: Receiver<EscrowAccounts>,
         timestamp_error_tolerance: Duration,
         receipt_max_value: u128,
     ) -> Vec<ReceiptCheck<TapReceipt>> {
         vec![
             Arc::new(AllocationEligible::new(indexer_allocations)),
-            Arc::new(SenderBalanceCheck::new(escrow_accounts)),
+            Arc::new(SenderBalanceCheck::new(
+                escrow_accounts_v1,
+                escrow_accounts_v2,
+            )),
             Arc::new(TimestampCheck::new(timestamp_error_tolerance)),
             Arc::new(DenyListCheck::new(pgpool.clone()).await),
             Arc::new(ReceiptMaxValueCheck::new(receipt_max_value)),

crates/service/src/tap.rs

@@ -13,30 +13,45 @@ use crate::{
 };
 
 pub struct SenderBalanceCheck {
-    escrow_accounts: Receiver<EscrowAccounts>,
+    escrow_accounts_v1: Receiver<EscrowAccounts>,
+    escrow_accounts_v2: Receiver<EscrowAccounts>,
 }
 
 impl SenderBalanceCheck {
-    pub fn new(escrow_accounts: Receiver<EscrowAccounts>) -> Self {
-        Self { escrow_accounts }
+    pub fn new(
+        escrow_accounts_v1: Receiver<EscrowAccounts>,
+        escrow_accounts_v2: Receiver<EscrowAccounts>,
+    ) -> Self {
+        Self {
+            escrow_accounts_v1,
+            escrow_accounts_v2,
+        }
     }
 }
 
 #[async_trait::async_trait]
 impl Check<TapReceipt> for SenderBalanceCheck {
-    async fn check(&self, ctx: &tap_core::receipt::Context, _: &CheckingReceipt) -> CheckResult {
-        let escrow_accounts_snapshot = self.escrow_accounts.borrow();
+    async fn check(
+        &self,
+        ctx: &tap_core::receipt::Context,
+        receipt: &CheckingReceipt,
+    ) -> CheckResult {
+        let escrow_accounts_snapshot_v1 = self.escrow_accounts_v1.borrow();
+        let escrow_accounts_snapshot_v2 = self.escrow_accounts_v2.borrow();
 
         let Sender(receipt_sender) = ctx
             .get::<Sender>()
             .ok_or(CheckError::Failed(anyhow::anyhow!("Could not find sender")))?;
 
+        // get balance for escrow account given receipt type
+        let balance_result = match receipt.signed_receipt() {
+            TapReceipt::V1(_) => escrow_accounts_snapshot_v1.get_balance_for_sender(receipt_sender),
+            TapReceipt::V2(_) => escrow_accounts_snapshot_v2.get_balance_for_sender(receipt_sender),
+        };
+
         // Check that the sender has a non-zero balance -- more advanced accounting is done in
         // `tap-agent`.
-        if !escrow_accounts_snapshot
-            .get_balance_for_sender(receipt_sender)
-            .is_ok_and(|balance| balance > U256::ZERO)
-        {
+        if !balance_result.is_ok_and(|balance| balance > U256::ZERO) {
             return Err(CheckError::Failed(anyhow!(
                 "Receipt sender `{}` does not have a sufficient balance",
                 receipt_sender,

crates/service/src/tap/checks/sender_balance_check.rs

@@ -90,7 +90,8 @@ async fn full_integration_test(database: PgPool) {
             receipts_verifier_address: test_assets::VERIFIER_ADDRESS,
         })
         .timestamp_buffer_secs(Duration::from_secs(10))
-        .escrow_accounts(escrow_accounts)
+        .escrow_accounts_v1(escrow_accounts.clone())
+        .escrow_accounts_v2(escrow_accounts)
         .dispute_manager(dispute_manager)
         .allocations(allocations)
         .build();