feat: grpc client for tap aggregator

crates/tap-agent/Cargo.toml

@@ -49,5 +49,6 @@ futures = { version = "0.3.30", default-features = false }
 [dev-dependencies]
 tempfile = "3.8.0"
 wiremock.workspace = true
+wiremock-grpc = "0.0.3-alpha3"
 test-assets = { path = "../test-assets" }
 test-log = { version = "0.2.12", default-features = false }

crates/tap-agent/src/agent/sender_account.rs

@@ -606,8 +606,7 @@ impl Actor for SenderAccount {
             .set(config.trigger_value as f64);
 
         let endpoint = Endpoint::new(sender_aggregator_endpoint.to_string())
-            .expect("Failed to create an endpoint for the sender aggregator")
-            .connect_timeout(config.rav_request_timeout);
+            .context("Failed to create an endpoint for the sender aggregator")?;
 
         let sender_aggregator = TapAggregatorClient::connect(endpoint.clone())
             .await
@@ -616,8 +615,11 @@ impl Actor for SenderAccount {
                     "Failed to connect to the TapAggregator endpoint '{}'",
                     endpoint.uri()
                 )
-            })?
-            .send_compressed(tonic::codec::CompressionEncoding::Zstd);
+            })?;
+        // wiremock_grpc used for tests doesn't support Zstd compression
+        #[cfg(not(test))]
+        let sender_aggregator =
+            sender_aggregator.send_compressed(tonic::codec::CompressionEncoding::Zstd);
         let state = State {
             prefix,
             sender_fee_tracker: SenderFeeTracker::new(config.rav_request_buffer),
@@ -1081,8 +1083,7 @@ pub mod tests {
         assert_not_triggered, assert_triggered,
         test::{
             actors::{create_mock_sender_allocation, MockSenderAllocation, TestableActor},
-            create_rav, start_test_aggregator_server, store_rav_with_options, INDEXER,
-            TAP_EIP712_DOMAIN_SEPARATOR,
+            create_rav, get_grpc_url, store_rav_with_options, INDEXER, TAP_EIP712_DOMAIN_SEPARATOR,
         },
     };
 
@@ -1205,10 +1206,6 @@ pub mod tests {
             .expect("Failed to update escrow_accounts channel");
 
         // Start a new mock aggregator server for this test
-        let (_server_handle, server_addr) = start_test_aggregator_server()
-            .await
-            .expect("Failed to start mock aggregator server");
-        let server_url = format!("http://{}", server_addr);
 
         let prefix = format!(
             "test-{}",
@@ -1224,7 +1221,7 @@ pub mod tests {
             escrow_subgraph,
             network_subgraph,
             domain_separator: TAP_EIP712_DOMAIN_SEPARATOR.clone(),
-            sender_aggregator_endpoint: Url::parse(&server_url).unwrap(),
+            sender_aggregator_endpoint: Url::parse(&get_grpc_url().await).unwrap(),
             allocation_ids: HashSet::new(),
             prefix: Some(prefix.clone()),
             retry_interval: RETRY_DURATION,

crates/tap-agent/src/agent/sender_accounts_manager.rs

@@ -609,7 +609,7 @@ mod tests {
     use reqwest::Url;
     use ruint::aliases::U256;
     use sqlx::{postgres::PgListener, PgPool};
-    use test_assets::{flush_messages, TAP_SENDER as SENDER};
+    use test_assets::{flush_messages, TAP_SENDER as SENDER, TAP_SIGNER as SIGNER};
     use thegraph_core::alloy::hex::ToHexExt;
     use tokio::sync::{mpsc, mpsc::error::TryRecvError, watch, Notify};
 
@@ -624,9 +624,8 @@ mod tests {
         },
         test::{
             actors::{DummyActor, MockSenderAccount, MockSenderAllocation, TestableActor},
-            create_rav, create_received_receipt, start_test_aggregator_server, store_rav,
-            store_receipt, ALLOCATION_ID_0, ALLOCATION_ID_1, INDEXER, SENDER_2, SIGNER,
-            TAP_EIP712_DOMAIN_SEPARATOR,
+            create_rav, create_received_receipt, get_grpc_url, store_rav, store_receipt,
+            ALLOCATION_ID_0, ALLOCATION_ID_1, INDEXER, SENDER_2, TAP_EIP712_DOMAIN_SEPARATOR,
         },
     };
 
@@ -672,19 +671,6 @@ mod tests {
         let (_, escrow_accounts_rx) = watch::channel(EscrowAccounts::default());
 
         // Start a new mock aggregator server for this test
-        let (server1_url, server2_url) = {
-            let (_server1_handle, server1_addr) = start_test_aggregator_server()
-                .await
-                .expect("Failed to start the first mock aggregator server");
-            let server1_url = format!("http://{}", server1_addr);
-
-            let (_server2_handle, server2_addr) = start_test_aggregator_server()
-                .await
-                .expect("Failed to start the second mock aggregator server");
-            let server2_url = format!("http://{}", server2_addr);
-
-            (server1_url, server2_url)
-        };
         let prefix = format!(
             "test-{}",
             PREFIX_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
@@ -698,8 +684,8 @@ mod tests {
             escrow_subgraph,
             network_subgraph,
             sender_aggregator_endpoints: HashMap::from([
-                (SENDER.1, Url::parse(&server1_url).unwrap()),
-                (SENDER_2.1, Url::parse(&server2_url).unwrap()),
+                (SENDER.1, Url::parse(&get_grpc_url().await).unwrap()),
+                (SENDER_2.1, Url::parse(&get_grpc_url().await).unwrap()),
             ]),
             prefix: Some(prefix.clone()),
         };
@@ -724,21 +710,6 @@ mod tests {
         let senders_to_signers = vec![(SENDER.1, vec![SIGNER.1])].into_iter().collect();
         let escrow_accounts = EscrowAccounts::new(HashMap::new(), senders_to_signers);
 
-        // Start a new mock aggregator server for this test
-        let (server1_url, server2_url) = {
-            let (_server1_handle, server1_addr) = start_test_aggregator_server()
-                .await
-                .expect("Failed to start the first mock aggregator server");
-            let server1_url = format!("http://{}", server1_addr);
-
-            let (_server2_handle, server2_addr) = start_test_aggregator_server()
-                .await
-                .expect("Failed to start the second mock aggregator server");
-            let server2_url = format!("http://{}", server2_addr);
-
-            (server1_url, server2_url)
-        };
-
         let prefix = format!(
             "test-{}",
             PREFIX_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
@@ -757,8 +728,8 @@ mod tests {
                 escrow_subgraph: get_subgraph_client().await,
                 network_subgraph: get_subgraph_client().await,
                 sender_aggregator_endpoints: HashMap::from([
-                    (SENDER.1, Url::parse(&server1_url).unwrap()),
-                    (SENDER_2.1, Url::parse(&server2_url).unwrap()),
+                    (SENDER.1, Url::parse(&get_grpc_url().await).unwrap()),
+                    (SENDER_2.1, Url::parse(&get_grpc_url().await).unwrap()),
                 ]),
                 prefix: Some(prefix),
             },