Merge branch 'main' into release-please--branches--main

Author: gusinacio

.sqlx/query-f1d8dcf24c9677ef789469d37733f354c5d246db5bb1506839cd7b07cdcc7065.json

@@ -76,12 +76,13 @@ 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"
 tonic = { version = "0.12.3", features = ["tls-roots", "gzip"] }
 prost = "0.13.4"
 prost-types = "0.13.3"
 tonic-build = "0.12.3"
 serde_yaml = "0.9.21"
+bon = "3.3"
+test-log = { version = "0.2.12", features = ["trace"] }
 
 [patch.crates-io.tap_core]
 git = "https://github.com/semiotic-ai/timeline-aggregation-protocol"

Cargo.lock

@@ -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),

Cargo.toml

@@ -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/escrow_accounts.rs

@@ -32,15 +32,17 @@ tracing-subscriber = { workspace = true, features = ["fmt"] }
 clap = { workspace = true, features = ["derive"] }
 build-info.workspace = true
 lazy_static.workspace = true
+prost.workspace = true 
 async-trait.workspace = true
 async-graphql = { version = "7.0.11", default-features = false }
 async-graphql-axum = "7.0.11"
 base64.workspace = true
 graphql = { git = "https://github.com/edgeandnode/toolshed", tag = "graphql-v0.3.0" }
 tap_core.workspace = true
 tap_graph.workspace = true
+tap_aggregator.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 = [
@@ -71,6 +73,7 @@ tower-service = "0.3.3"
 tokio-test = "0.4.4"
 wiremock.workspace = true
 insta = "1.41.1"
+test-log.workspace = true
 
 [build-dependencies]
 build-info-build = { version = "0.0.39", default-features = false }

crates/monitor/src/lib.rs

@@ -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/Cargo.toml

@@ -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,
             };