Merge pull request #3 from weaveVM/low-latency

feat: v0.1.1

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "exex-avs-operator"
 edition = "2021"
-version = "0.1.0"
+version = "0.1.1"
 rust-version = "1.82"
 license = "MIT"
 homepage = "https://docs.wvm.dev"

README.md

@@ -8,10 +8,9 @@
 Operators-As-ExEx is a paradigm that integrates partially or completely the operator of an Actively Validated Services as [Execution Extensions](https://exex.rs) in Reth. This approach aligns perfectly with events-driven activated predefined-actions (EDAs) such as Keeper Networks.
 
 ## About This ExEx
-This ExEx is the re-implementation of Eigenlayer's [hello-world-avs](https://github.com/Layr-Labs/hello-world-avs) as an ExEx. this work comes to demonstrate in practice how the Operator-As-ExEx paradigm works.
+This ExEx is the re-implementation of Eigenlayer's [hello-world-avs](https://github.com/Layr-Labs/hello-world-avs) as an ExEx. this work comes to demonstrate in practice how the Operator-As-ExEx paradigm works. This ExEx is for demoing and educational purposes only, ***DO NOT USE IT IN PRODUCTION.***
 
 ## Low Latency EDA
-
 This ExEx shows how to build a high-performance EDA-AVS-Operator by minimizing network latency. Rather than continuously polling JSON-RPC endpoints for AVS Service Manager smart contract events or monitoring blockchain events, this implementation uses direct ExEx notifications that has shared memory communication with the Reth node.
 
 The operator initializes the RPC provider just once, using it only when submitting task results. All other monitoring/polling - whether for new AVS tasks or EDA-task completion events - happens through ExEx notifications instead of current polling methods in AVS operators.

src/main.rs

@@ -30,21 +30,22 @@ async fn exex_operator<Node: FullNodeComponents>(mut ctx: ExExContext<Node>) ->
         };
 
         if let Some(committed_chain) = notification.committed_chain() {
-            // Get all block's receipts
-            let last_block_transactions_receipts = committed_chain
-                .blocks_and_receipts()
-                // Flat map the (block, receipts) tuple to just receipts
-                .flat_map(|(_block, receipts)| receipts.iter().cloned())
-                .collect();
+            // Get all blocks and receipts from this commitment
+            let blocks_transactions_receipts: Vec<_> =
+                committed_chain.blocks_and_receipts().collect();
 
-            // monitor the AVS Service Manager tasks
-            monitor_new_tasks_of_block(
-                provider.clone(),
-                avs_manager,
-                last_block_transactions_receipts,
-            )
-            .await
-            .unwrap();
+            // Processing all the block(s) of this chain commitment (block(s) batch)
+            for (block, receipts) in blocks_transactions_receipts {
+                // monitor the AVS Service Manager tasks for each block
+                monitor_new_tasks_of_block(
+                    provider.clone(),
+                    avs_manager.clone(),
+                    receipts.iter().cloned().collect(),
+                    block,
+                )
+                .await
+                .unwrap();
+            }
             ctx.events
                 .send(ExExEvent::FinishedHeight(committed_chain.tip().num_hash()))?;
         }

src/utils/monitor_tasks.rs

@@ -8,7 +8,7 @@ use ethers::signers::LocalWallet;
 use ethers::utils::keccak256;
 use eyre::Result;
 use once_cell::sync::Lazy;
-use reth::primitives::{Log, Receipt};
+use reth::primitives::{Log, Receipt, SealedBlockWithSenders};
 use std::{env, str::FromStr, sync::Arc};
 use zerocopy::IntoBytes;
 
@@ -92,12 +92,14 @@ pub async fn decode_new_task_created_event(
 pub async fn monitor_new_tasks_of_block(
     provider: Provider<Http>,
     contract_address: Address,
-    last_block_transactions_receipts: Vec<Option<Receipt>>,
+    current_block_transactions_receipts: Vec<Option<Receipt>>,
+    current_sealed_block_with_senders: &SealedBlockWithSenders,
 ) -> Result<()> {
+    // now you can use current_sealed_block_with_senders for EDA task scanning purposes
     let event_signature = H256::from_str(NEW_TASK_CREATED_EVENT_NAME)?;
 
     let mut filtered_logs: Vec<Log> = Vec::new();
-    for receipt in last_block_transactions_receipts.into_iter().flatten() {
+    for receipt in current_block_transactions_receipts.into_iter().flatten() {
         for log in receipt.logs {
             if log.address == Address::from(contract_address).as_bytes()
                 && !log.topics().is_empty()