feat!: Refactor EventFlows for EVM and Remove EventWatchers (#423)

* Improve intuition behind TangleEventListener

* Get tangle and eigen integration tests passing - again

* Few compile errors left

* chore: Delete lots of old code

* feat: Port EVM/event-watcher to EVM/event-flows

* chore: cleanup

* chore: Use improved target pathing

* chore: ensure tests run

* chore: Remove use of traits for allowing specific values

blueprint-manager/src/sources/mod.rs

@@ -150,14 +150,12 @@ pub fn generate_process_arguments(
     }
 
     arguments.extend([
-        format!("--bind-addr={}", gadget_config.bind_addr),
-        format!("--bind-port={}", gadget_config.bind_port),
+        format!("--target-addr={}", gadget_config.bind_addr),
+        format!("--target-port={}", gadget_config.bind_port),
         format!("--http-rpc-url={}", gadget_config.http_rpc_url),
         format!("--ws-rpc-url={}", gadget_config.ws_rpc_url),
         format!("--keystore-uri={}", gadget_config.keystore_uri),
         format!("--chain={}", gadget_config.chain),
-        format!("--blueprint-id={}", blueprint_id),
-        format!("--service-id={}", service_id),
         format!("--protocol={}", protocol),
         format!(
             "--log-id=Blueprint-{blueprint_id}-Service-{service_id}-{}",

blueprint-manager/src/sources/testing.rs

@@ -32,15 +32,36 @@ impl BinarySourceFetcher for TestSourceFetcher {
             "release"
         };
         let base_path = std::path::absolute(git_repo_root.join(&base_path_str))?;
-        let binary_path = git_repo_root
-            .join(&base_path)
-            .join("target")
-            .join(profile)
-            .join(&cargo_bin);
+
+        let target_dir = match std::env::var("CARGO_TARGET_DIR") {
+            Ok(target) => PathBuf::from(target),
+            Err(_) => git_repo_root.join(&base_path).join("target"),
+        };
+
+        let binary_path = target_dir.join(profile).join(&cargo_bin);
         let binary_path = std::path::absolute(&binary_path)?;
 
         trace!("Base Path: {}", base_path.display());
         trace!("Binary Path: {}", binary_path.display());
+
+        // Run cargo build on the cargo_bin and ensure it build to the binary_path
+        let mut command = tokio::process::Command::new("cargo");
+        command
+            .arg("build")
+            .arg(format!("--target-dir={}", target_dir.display()))
+            .arg("--bin")
+            .arg(&cargo_bin);
+
+        if !cfg!(debug_assertions) {
+            command.arg("--release");
+        }
+
+        let output = command.current_dir(&base_path).output().await?;
+
+        if !output.status.success() {
+            return Err(Report::msg(format!("Failed to build binary: {:?}", output)));
+        }
+
         Ok(binary_path)
     }
 

blueprint-test-utils/Cargo.toml

@@ -52,7 +52,6 @@ alloy-transport = { workspace = true }
 
 # Utility
 testcontainers = { workspace = true }
-tempfile = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }
 
 [dev-dependencies]

blueprint-test-utils/src/helpers.rs

@@ -89,8 +89,8 @@ impl BlueprintProcessManager {
         ws_endpoint: &str,
         protocol: Protocol,
     ) -> Result<BlueprintProcess, std::io::Error> {
-        let tmp_dir = tempfile::TempDir::new()?;
-        let keystore_uri = tmp_dir.path().join(format!(
+        let tmp_dir = std::env::current_dir()?;
+        let keystore_uri = tmp_dir.join(format!(
             "keystores/{}/{}",
             NAME_IDS[instance_id].to_lowercase(),
             uuid::Uuid::new_v4()
@@ -134,8 +134,8 @@ impl BlueprintProcessManager {
 
         let mut arguments = vec![
             "run".to_string(),
-            format!("--bind-addr={}", IpAddr::from_str("127.0.0.1").unwrap()),
-            format!("--bind-port={}", find_open_tcp_bind_port()),
+            format!("--target-addr={}", IpAddr::from_str("127.0.0.1").unwrap()),
+            format!("--target-port={}", find_open_tcp_bind_port()),
             format!("--http-rpc-url={}", Url::parse(http_endpoint).unwrap()),
             format!("--ws-rpc-url={}", Url::parse(ws_endpoint).unwrap()),
             format!("--keystore-uri={}", keystore_uri_str.clone()),
@@ -169,6 +169,10 @@ impl BlueprintProcessManager {
                     "--strategy-manager={}",
                     EigenlayerContractAddresses::default().strategy_manager_address
                 ));
+                arguments.push(format!(
+                    "--avs-directory={}",
+                    EigenlayerContractAddresses::default().avs_directory_address
+                ));
             }
             Protocol::Symbiotic => {
                 arguments.push(format!(

blueprints/incredible-squaring-eigenlayer/README.md

@@ -36,5 +36,5 @@ cargo build --release -p incredible-squaring-blueprint-eigenlayer
 - We have a test for running this AVS Blueprint on a local Anvil Testnet. You can run the test with the following:
 
 ```bash
-RUST_LOG=gadget=trace cargo test --package incredible-squaring-blueprint-eigenlayer test_eigenlayer_incredible_squaring_blueprint -- --nocapture
+RUST_LOG=gadget=trace cargo test --package blueprint-test-utils test_eigenlayer_incredible_squaring_blueprint -- --nocapture
 ```

blueprints/incredible-squaring-eigenlayer/src/jobs/compute_x_square.rs

@@ -9,6 +9,7 @@ use ark_ff::{BigInteger, PrimeField};
 use color_eyre::Result;
 use eigensdk::crypto_bls::BlsKeyPair;
 use eigensdk::crypto_bls::OperatorId;
+use gadget_sdk::event_listener::evm::contracts::EvmContractEventListener;
 use gadget_sdk::{error, info, job};
 use std::{convert::Infallible, ops::Deref};
 use IncredibleSquaringTaskManager::TaskResponse;
@@ -17,15 +18,11 @@ use IncredibleSquaringTaskManager::TaskResponse;
 #[job(
     id = 0,
     params(number_to_be_squared, task_created_block, quorum_numbers, quorum_threshold_percentage, task_index),
-    result(_),
     event_listener(
-        listener = EvmContractEventListener(
-            instance = IncredibleSquaringTaskManager,
-            abi = INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
-        ),
-        event = IncredibleSquaringTaskManager::NewTaskCreated,
+        listener = EvmContractEventListener<IncredibleSquaringTaskManager::NewTaskCreated>,
+        instance = IncredibleSquaringTaskManager,
+        abi = INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
         pre_processor = convert_event_to_inputs,
-        post_processor = noop,
     ),
 )]
 pub async fn xsquare_eigen(
@@ -35,7 +32,7 @@ pub async fn xsquare_eigen(
     quorum_numbers: Bytes,
     quorum_threshold_percentage: u8,
     task_index: u32,
-) -> Result<u32, Infallible> {
+) -> std::result::Result<u32, Infallible> {
     // Calculate our response to job
     let task_response = TaskResponse {
         referenceTaskIndex: task_index,
@@ -79,22 +76,24 @@ pub async fn xsquare_eigen(
 /// Uses a tuple to represent the return type because
 /// the macro will index all values in the #[job] function
 /// and parse the return type by the index.
-pub fn convert_event_to_inputs(
-    event: IncredibleSquaringTaskManager::NewTaskCreated,
-    _index: u32,
-) -> (U256, u32, Bytes, u8, u32) {
+pub async fn convert_event_to_inputs(
+    (event, _log): (
+        IncredibleSquaringTaskManager::NewTaskCreated,
+        alloy_rpc_types::Log,
+    ),
+) -> Result<(U256, u32, Bytes, u8, u32), gadget_sdk::Error> {
     let task_index = event.taskIndex;
     let number_to_be_squared = event.task.numberToBeSquared;
     let task_created_block = event.task.taskCreatedBlock;
     let quorum_numbers = event.task.quorumNumbers;
     let quorum_threshold_percentage = event.task.quorumThresholdPercentage.try_into().unwrap();
-    (
+    Ok((
         number_to_be_squared,
         task_created_block,
         quorum_numbers,
         quorum_threshold_percentage,
         task_index,
-    )
+    ))
 }
 
 /// Helper for converting a PrimeField to its U256 representation for Ethereum compatibility

blueprints/incredible-squaring-eigenlayer/src/jobs/initialize_task.rs

@@ -2,6 +2,7 @@ use crate::{
     contexts::aggregator::AggregatorContext, IncredibleSquaringTaskManager,
     INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
 };
+use gadget_sdk::event_listener::evm::contracts::EvmContractEventListener;
 use gadget_sdk::{info, job};
 use std::{convert::Infallible, ops::Deref};
 use IncredibleSquaringTaskManager::Task;
@@ -13,15 +14,11 @@ const BLOCK_TIME_SECONDS: u32 = 12;
 #[job(
     id = 1,
     params(task, task_index),
-    result(_),
     event_listener(
-        listener = EvmContractEventListener(
-            instance = IncredibleSquaringTaskManager,
-            abi = INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
-        ),
-        event = IncredibleSquaringTaskManager::NewTaskCreated,
+        listener = EvmContractEventListener<IncredibleSquaringTaskManager::NewTaskCreated>,
+        instance = IncredibleSquaringTaskManager,
+        abi = INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
         pre_processor = convert_event_to_inputs,
-        post_processor = noop,
     ),
 )]
 pub async fn initialize_bls_task(
@@ -59,10 +56,12 @@ pub async fn initialize_bls_task(
 /// Uses a tuple to represent the return type because
 /// the macro will index all values in the #[job] function
 /// and parse the return type by the index.
-pub fn convert_event_to_inputs(
-    event: IncredibleSquaringTaskManager::NewTaskCreated,
-    _index: u32,
-) -> (Task, u32) {
-    let task_index = event.taskIndex;
-    (event.task, task_index)
+pub async fn convert_event_to_inputs(
+    event: (
+        IncredibleSquaringTaskManager::NewTaskCreated,
+        alloy_rpc_types::Log,
+    ),
+) -> Result<(Task, u32), gadget_sdk::Error> {
+    let task_index = event.0.taskIndex;
+    Ok((event.0.task, task_index))
 }

blueprints/incredible-squaring-eigenlayer/src/lib.rs

@@ -6,7 +6,6 @@ use serde::{Deserialize, Serialize};
 pub mod constants;
 pub mod contexts;
 pub mod jobs;
-// pub mod runner;
 
 sol!(
     #[allow(missing_docs)]
@@ -20,7 +19,3 @@ load_abi!(
     INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
     "contracts/out/IncredibleSquaringTaskManager.sol/IncredibleSquaringTaskManager.json"
 );
-
-pub fn noop(_: u32) {
-    // This function intentionally does nothing
-}

blueprints/incredible-squaring-eigenlayer/src/main.rs

@@ -1,9 +1,8 @@
 use alloy_network::EthereumWallet;
-use alloy_provider::ProviderBuilder;
 use alloy_signer_local::PrivateKeySigner;
 use color_eyre::Result;
+use gadget_sdk::utils::evm::get_wallet_provider_http;
 use gadget_sdk::{
-    events_watcher::evm::DefaultNodeConfig,
     info,
     runners::{eigenlayer::EigenlayerConfig, BlueprintRunner},
 };
@@ -23,10 +22,8 @@ async fn main() {
         .parse()
         .expect("failed to generate wallet ");
     let wallet = EthereumWallet::from(signer);
-    let provider = ProviderBuilder::new()
-        .with_recommended_fillers()
-        .wallet(wallet.clone())
-        .on_http(env.http_rpc_endpoint.parse()?);
+    let provider = get_wallet_provider_http(&env.http_rpc_endpoint, wallet.clone());
+
     info!("Task Manager Address: {:?}", *TASK_MANAGER_ADDRESS);
     let contract = IncredibleSquaringTaskManager::IncredibleSquaringTaskManagerInstance::new(
         *TASK_MANAGER_ADDRESS,
@@ -63,7 +60,7 @@ async fn main() {
 
     let server_address = format!("{}:{}", env.target_addr, 8081);
     let aggregator_client = AggregatorClient::new(&server_address)?;
-    let x_square_eigen = XsquareEigenEventHandler::<DefaultNodeConfig> {
+    let x_square_eigen = XsquareEigenEventHandler {
         ctx: aggregator_client,
         contract: contract.clone(),
         contract_instance: Default::default(),
@@ -79,7 +76,7 @@ async fn main() {
     .await
     .unwrap();
 
-    let initialize_task = InitializeBlsTaskEventHandler::<DefaultNodeConfig> {
+    let initialize_task = InitializeBlsTaskEventHandler {
         ctx: aggregator_context.clone(),
         contract,
         contract_instance: Default::default(),

blueprints/incredible-squaring-symbiotic/src/lib.rs

@@ -1,5 +1,6 @@
 use alloy_primitives::U256;
 use alloy_sol_types::sol;
+use gadget_sdk::event_listener::evm::contracts::EvmContractEventListener;
 use gadget_sdk::{job, load_abi};
 use serde::{Deserialize, Serialize};
 use std::ops::Deref;
@@ -17,36 +18,30 @@ load_abi!(
     "contracts/out/IncredibleSquaringTaskManager.sol/IncredibleSquaringTaskManager.json"
 );
 
-pub fn noop(_: U256) {
-    // This function intentionally does nothing
-}
-
 #[derive(Clone)]
 pub struct MyContext;
 
 /// Returns x^2 saturating to [`u64::MAX`] if overflow occurs.
 #[job(
     id = 0,
     params(x),
-    result(_),
     event_listener(
-        listener = EvmContractEventListener(
-            instance = IncredibleSquaringTaskManager,
-            abi = INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
-        ),
-        event = IncredibleSquaringTaskManager::NewTaskCreated,
+        listener = EvmContractEventListener<IncredibleSquaringTaskManager::NewTaskCreated>,
+        instance = IncredibleSquaringTaskManager,
+        abi = INCREDIBLE_SQUARING_TASK_MANAGER_ABI_STRING,
         pre_processor = convert_event_to_inputs,
-        post_processor = noop,
     ),
 )]
-pub fn xsquare(context: MyContext, x: U256) -> Result<U256, gadget_sdk::Error> {
+pub fn xsquare(x: U256, context: MyContext) -> Result<U256, gadget_sdk::Error> {
     Ok(x.saturating_pow(U256::from(2)))
 }
 
 /// Converts the event to inputs.
-pub fn convert_event_to_inputs(
-    event: IncredibleSquaringTaskManager::NewTaskCreated,
-    _i: u32,
-) -> (U256,) {
-    (event.task.numberToBeSquared,)
+pub async fn convert_event_to_inputs(
+    (event, _log): (
+        IncredibleSquaringTaskManager::NewTaskCreated,
+        alloy_rpc_types::Log,
+    ),
+) -> Result<(U256,), gadget_sdk::Error> {
+    Ok((event.task.numberToBeSquared,))
 }

blueprints/incredible-squaring-symbiotic/src/main.rs

@@ -1,13 +1,12 @@
 use alloy_network::EthereumWallet;
-use alloy_provider::ProviderBuilder;
 use color_eyre::Result;
-use gadget_sdk::events_watcher::evm::DefaultNodeConfig;
 use gadget_sdk::runners::symbiotic::SymbioticConfig;
 use gadget_sdk::runners::BlueprintRunner;
 use gadget_sdk::{info, keystore::BackendExt};
 use incredible_squaring_blueprint_symbiotic::{self as blueprint, IncredibleSquaringTaskManager};
 
 use alloy_primitives::{address, Address};
+use gadget_sdk::utils::evm::get_wallet_provider_http;
 use lazy_static::lazy_static;
 use std::env;
 
@@ -24,24 +23,21 @@ async fn main() {
     let operator_signer = env.keystore()?.ecdsa_key()?.alloy_key()?;
     let wallet = EthereumWallet::new(operator_signer);
 
-    let provider = ProviderBuilder::new()
-        .with_recommended_fillers()
-        .wallet(wallet.clone())
-        .on_http(env.http_rpc_endpoint.parse()?);
+    let provider = get_wallet_provider_http(&env.http_rpc_endpoint, wallet);
 
     let contract = IncredibleSquaringTaskManager::IncredibleSquaringTaskManagerInstance::new(
         *TASK_MANAGER_ADDRESS,
         provider,
     );
 
-    let x_square = blueprint::XsquareEventHandler::<DefaultNodeConfig> {
+    let x_square = blueprint::XsquareEventHandler {
         context: blueprint::MyContext {},
         contract: contract.clone(),
         contract_instance: Default::default(),
     };
 
     info!("~~~ Executing the incredible squaring blueprint ~~~");
-    let symb_config = SymbioticConfig {};
+    let symb_config = SymbioticConfig::default();
     BlueprintRunner::new(symb_config, env)
         .job(x_square)
         .run()