Use transaction watcher in tests

crates/chainio/clients/elcontracts/src/writer.rs

@@ -200,6 +200,7 @@ mod tests {
             self, get_delegation_manager_address, get_erc20_mock_strategy,
             get_service_manager_address, get_strategy_manager_address,
         },
+        transaction::wait_transaction,
     };
     use eigen_types::operator::Operator;
     use eigen_utils::{
@@ -329,9 +330,8 @@ mod tests {
             .register_as_operator(operator)
             .await
             .unwrap();
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
-        // this sleep is needed so that we wait for the tx to be processed
-        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
         let receipt = provider.get_transaction_receipt(tx_hash).await.unwrap();
         assert!(receipt.unwrap().status());
 
@@ -353,22 +353,20 @@ mod tests {
             .update_operator_details(operator_modified)
             .await
             .unwrap();
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
-        // this sleep is needed so that we wait for the tx to be processed
-        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
         let receipt = provider.get_transaction_receipt(tx_hash).await.unwrap();
         assert!(receipt.unwrap().status());
 
         // Third test: deposit_erc20_into_strategy
         let amount = U256::from_str("100").unwrap();
-        let strategy_addr = get_erc20_mock_strategy(http_endpoint).await;
+        let strategy_addr = get_erc20_mock_strategy(http_endpoint.clone()).await;
         let tx_hash = el_chain_writer
             .deposit_erc20_into_strategy(strategy_addr, amount)
             .await
             .unwrap();
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
-        // this sleep is needed so that we wait for the tx to be processed
-        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
         let receipt = provider.get_transaction_receipt(tx_hash).await.unwrap();
         assert!(receipt.unwrap().status());
     }

crates/services/bls_aggregation/src/bls_agg.rs

@@ -152,7 +152,7 @@ impl<A: AvsRegistryService + Send + Sync + Clone + 'static> BlsAggregatorService
         Ok(())
     }
 
-    /// Processs signatures received from the channel and sends
+    /// Processes signatures received from the channel and sends
     /// the signed task response to the task channel.
     ///
     /// # Arguments

crates/services/bls_aggregation/src/bls_agg_test.rs

@@ -19,6 +19,7 @@ pub mod integration_test {
             get_registry_coordinator_address, get_service_manager_address,
         },
         test_data::TestData,
+        transaction::wait_transaction,
     };
     use eigen_types::{
         avs::TaskIndex,
@@ -34,7 +35,7 @@ pub mod integration_test {
     use serde::Deserialize;
     use sha2::{Digest, Sha256};
     use std::time::Duration;
-    use tokio::{task, time::sleep};
+    use tokio::task;
     use tokio_util::sync::CancellationToken;
 
     const PRIVATE_KEY_1: &str = "ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"; // the owner addr
@@ -164,7 +165,7 @@ pub mod integration_test {
             .unwrap();
 
         // Register operator
-        avs_writer
+        let tx_hash = avs_writer
             .register_operator_in_quorum_with_avs_registry_coordinator(
                 bls_key_pair.clone(),
                 salt,
@@ -174,9 +175,7 @@ pub mod integration_test {
             )
             .await
             .unwrap();
-
-        // Sleep is needed so registered operators are accesible to the OperatorInfoServiceInMemory
-        sleep(Duration::from_secs(3)).await;
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
         // Create aggregation service
         let avs_registry_service =
@@ -332,14 +331,14 @@ pub mod integration_test {
 
         let avs_writer = AvsRegistryChainWriter::build_avs_registry_chain_writer(
             get_test_logger(),
-            http_endpoint,
+            http_endpoint.clone(),
             PRIVATE_KEY_2.to_string(),
             registry_coordinator_address,
             operator_state_retriever_address,
         )
         .await
         .unwrap();
-        avs_writer
+        let tx_hash = avs_writer
             .register_operator_in_quorum_with_avs_registry_coordinator(
                 bls_key_pair_2.clone(),
                 salt,
@@ -349,9 +348,7 @@ pub mod integration_test {
             )
             .await
             .unwrap();
-
-        // Sleep is needed so registered operators are accesible to the OperatorInfoServiceInMemory
-        sleep(Duration::from_secs(3)).await;
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
         // Create aggregation service
         let avs_registry_service =
@@ -521,7 +518,7 @@ pub mod integration_test {
         task::spawn(async move { operators_info_clone.start_service(&token_clone, 0, 0).await });
 
         // Register operator
-        avs_writer
+        let tx_hash = avs_writer
             .register_operator_in_quorum_with_avs_registry_coordinator(
                 bls_key_pair_1.clone(),
                 salt,
@@ -531,18 +528,20 @@ pub mod integration_test {
             )
             .await
             .unwrap();
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
         let avs_writer = AvsRegistryChainWriter::build_avs_registry_chain_writer(
             // TODO: check if needed
             get_test_logger(),
-            http_endpoint,
+            http_endpoint.clone(),
             PRIVATE_KEY_2.to_string(),
             registry_coordinator_address,
             operator_state_retriever_address,
         )
         .await
         .unwrap();
-        avs_writer
+
+        let tx_hash = avs_writer
             .register_operator_in_quorum_with_avs_registry_coordinator(
                 bls_key_pair_2.clone(),
                 salt,
@@ -552,9 +551,7 @@ pub mod integration_test {
             )
             .await
             .unwrap();
-
-        // Sleep is needed so registered operators are accesible to the OperatorInfoServiceInMemory
-        sleep(Duration::from_secs(3)).await;
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
 
         // Create aggregation service
         let avs_registry_service =

crates/services/operatorsinfo/src/operatorsinfo_inmemory.rs

@@ -296,6 +296,7 @@ mod tests {
         get_operator_state_retriever_address, get_registry_coordinator_address,
         get_strategy_manager_address,
     };
+    use eigen_testing_utils::transaction::wait_transaction;
     use eigen_types::operator::Operator;
     use eigen_utils::get_provider;
     use std::str::FromStr;
@@ -389,9 +390,6 @@ mod tests {
         )
         .await;
 
-        // need to wait at least 3 seconds to get the event processed
-        tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
-
         cancel_token.clone().cancel();
 
         let address = address!("f39fd6e51aad88f6f4ce6ab8827279cfffb92266");
@@ -448,7 +446,6 @@ mod tests {
             "8949062771264691130193054363356855357736539613420316273398900351143637925935",
         )
         .await;
-        tokio::time::sleep(tokio::time::Duration::from_secs(3)).await; // need to wait atleast 3 second to get the event processed
 
         cancel_token.clone().cancel();
 
@@ -528,7 +525,7 @@ mod tests {
         let quorum_numbers = Bytes::from_str("0x00").unwrap();
         let socket = "socket";
 
-        let _ = avs_registry_writer
+        let tx_hash = avs_registry_writer
             .register_operator_in_quorum_with_avs_registry_coordinator(
                 bls_key_pair,
                 salt,
@@ -538,5 +535,6 @@ mod tests {
             )
             .await
             .unwrap();
+        wait_transaction(&http_endpoint, tx_hash).await.unwrap();
     }
 }

testing/testing-utils/Cargo.toml

@@ -15,10 +15,11 @@ workspace = true
 alloy-network.workspace = true
 alloy-primitives.workspace = true
 alloy-provider.workspace = true
+alloy-transport.workspace = true
 alloy-transport-http.workspace = true
 eigen-utils.workspace = true
 once_cell.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-tokio.workspace = true
 testcontainers.workspace = true
+url.workspace = true

testing/testing-utils/src/transaction.rs

@@ -1,25 +1,45 @@
-use alloy_primitives::FixedBytes;
-use alloy_provider::Provider;
+use alloy_primitives::{FixedBytes, TxHash};
+use alloy_provider::{PendingTransactionBuilder, Provider, ProviderBuilder};
+use alloy_transport::{TransportErrorKind, TransportResult};
 use eigen_utils::get_provider;
-use tokio::time::{sleep, Duration};
+use url::Url;
 
 /// Retrieves the status of a transaction from its hash.
 ///
 /// # Arguments
 ///
+/// `rpc_url` - The RPC URL.
 /// `tx_hash` - The hash of the transaction.
 ///
 /// # Returns
 ///
 /// A bool indicating wether the transaction was successful or not.
 pub async fn get_transaction_status(rpc_url: String, tx_hash: FixedBytes<32>) -> bool {
-    // this sleep is needed so that we wait for the tx to be processed
-    sleep(Duration::from_millis(500)).await;
+    if wait_transaction(&rpc_url, tx_hash).await.is_err() {
+        return false;
+    }
+
     get_provider(&rpc_url)
-        .clone()
         .get_transaction_receipt(tx_hash)
         .await
         .unwrap_or(None)
         .map(|receipt| receipt.status())
         .unwrap_or(false)
 }
+
+/// Wait for a transaction to finish.
+///
+/// # Arguments
+///
+/// `rpc_url` - The RPC URL.
+/// `tx_hash` - The hash of the transaction.
+///
+/// # Returns
+///
+/// A [`TransportResult`] containing the transaction hash.
+pub async fn wait_transaction(rpc_url: &str, tx_hash: FixedBytes<32>) -> TransportResult<TxHash> {
+    let url = Url::parse(rpc_url).map_err(|_| TransportErrorKind::custom_str("Invalid RPC URL"))?;
+    let root_provider = ProviderBuilder::new().on_http(url);
+    let pending_tx = PendingTransactionBuilder::new(&root_provider, tx_hash);
+    pending_tx.watch().await
+}