Merge branch 'staging' into feat/aggregation-mode-risc0

Author: MarcosNicolau

.gitignore

@@ -12,6 +12,7 @@ batcher/aligned/batch_inclusion_responses/*
 **/broadcast
 volume
 config-files/*.last_processed_batch.json
+config-files/*.last_aggregated_block.json
 
 nonce_*.bin
 

Makefile

@@ -9,7 +9,7 @@ AGG_CONFIG_FILE?=config-files/config-aggregator.yaml
 
 OPERATOR_VERSION=v0.15.1
 EIGEN_SDK_GO_VERSION_TESTNET=v0.2.0-beta.1
-EIGEN_SDK_GO_VERSION_MAINNET=v0.1.13
+EIGEN_SDK_GO_VERSION_MAINNET=v0.2.0-beta.1
 
 ifeq ($(OS),Linux)
 	BUILD_ALL_FFI = $(MAKE) build_all_ffi_linux

aggregation_mode/Cargo.lock

@@ -1,22 +1,26 @@
-use serde::Deserialize;
-use std::{fs::File, io::Read};
+use serde::{Deserialize, Serialize};
+use std::{fs::File, fs::OpenOptions, io::Read, io::Write};
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct ECDSAConfig {
     pub private_key_store_path: String,
     pub private_key_store_password: String,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Serialize)]
+pub struct LastAggregatedBlock {
+    pub last_aggregated_block: u64,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
 pub struct Config {
     pub eth_rpc_url: String,
     pub eth_ws_url: String,
     pub max_proofs_in_queue: u16,
     pub proof_aggregation_service_address: String,
     pub aligned_service_manager_address: String,
+    pub last_aggregated_block_filepath: String,
     pub ecdsa: ECDSAConfig,
-    pub fetch_logs_from_secs_ago: u64,
-    pub block_time_secs: u64,
 }
 
 impl Config {
@@ -27,4 +31,32 @@ impl Config {
         let config: Config = serde_yaml::from_str(&contents)?;
         Ok(config)
     }
+
+    pub fn get_last_aggregated_block(&self) -> Result<u64, Box<dyn std::error::Error>> {
+        let mut file = File::open(&self.last_aggregated_block_filepath)?;
+        let mut contents = String::new();
+        file.read_to_string(&mut contents)?;
+        let lab: LastAggregatedBlock = serde_json::from_str(&contents)?;
+        Ok(lab.last_aggregated_block)
+    }
+
+    pub fn update_last_aggregated_block(
+        &self,
+        last_aggregated_block: u64,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let last_aggregated_block_struct = LastAggregatedBlock {
+            last_aggregated_block,
+        };
+
+        let mut file = OpenOptions::new()
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(&self.last_aggregated_block_filepath)?;
+
+        let content = serde_json::to_string(&last_aggregated_block_struct)?;
+        file.write_all(content.as_bytes())?;
+
+        Ok(())
+    }
 }

aggregation_mode/src/backend/config.rs

@@ -28,8 +28,7 @@ pub enum ProofsFetcherError {
 pub struct ProofsFetcher {
     rpc_provider: RPCProvider,
     aligned_service_manager: AlignedLayerServiceManagerContract,
-    fetch_from_secs_ago: u64,
-    block_time_secs: u64,
+    last_aggregated_block: u64,
 }
 
 impl ProofsFetcher {
@@ -42,31 +41,52 @@ impl ProofsFetcher {
             rpc_provider.clone(),
         );
 
+        let last_aggregated_block = config.get_last_aggregated_block().unwrap();
+
         Self {
             rpc_provider,
             aligned_service_manager,
-            fetch_from_secs_ago: config.fetch_logs_from_secs_ago,
-            block_time_secs: config.block_time_secs,
+            last_aggregated_block,
         }
     }
 
-    pub async fn fetch(&self, engine: ZKVMEngine) -> Result<Vec<AlignedProof>, ProofsFetcherError> {
-        let from_block = self.get_block_number_to_fetch_from().await?;
+    pub async fn fetch(
+        &mut self,
+        engine: ZKVMEngine,
+    ) -> Result<Vec<AlignedProof>, ProofsFetcherError> {
+        // Get current block
+        let current_block = self
+            .rpc_provider
+            .get_block_number()
+            .await
+            .map_err(|e| ProofsFetcherError::GetBlockNumber(e.to_string()))?;
+
+        if current_block < self.last_aggregated_block {
+            return Err(ProofsFetcherError::GetBlockNumber(
+                "Invalid last processed block".to_string(),
+            ));
+        }
+
         info!(
-            "Fetching proofs from batch logs starting from block number {}",
-            from_block
+            "Fetching proofs from batch logs starting from block number {} upto {}",
+            self.last_aggregated_block, current_block
         );
+
         // Subscribe to NewBatch event from AlignedServiceManager
         let logs = self
             .aligned_service_manager
             .NewBatchV3_filter()
-            .from_block(from_block)
+            .from_block(self.last_aggregated_block)
+            .to_block(current_block)
             .query()
             .await
             .map_err(|e| ProofsFetcherError::GetLogs(e.to_string()))?;
 
         info!("Logs collected {}", logs.len());
 
+        // Update last processed block after collecting logs
+        self.last_aggregated_block = current_block;
+
         let mut proofs = vec![];
 
         for (batch, _) in logs {
@@ -145,15 +165,7 @@ impl ProofsFetcher {
         Ok(proofs)
     }
 
-    async fn get_block_number_to_fetch_from(&self) -> Result<u64, ProofsFetcherError> {
-        let block_number = self
-            .rpc_provider
-            .get_block_number()
-            .await
-            .map_err(|e| ProofsFetcherError::GetBlockNumber(e.to_string()))?;
-
-        let number_of_blocks_in_the_past = self.fetch_from_secs_ago / self.block_time_secs;
-
-        Ok(block_number.saturating_sub(number_of_blocks_in_the_past))
+    pub fn get_last_aggregated_block(&self) -> u64 {
+        self.last_aggregated_block
     }
 }

aggregation_mode/src/backend/fetcher.rs

@@ -74,14 +74,17 @@ impl ProofAggregator {
         }
     }
 
-    pub async fn start(&mut self) {
+    pub async fn start(&mut self, config: &Config) {
         info!("Starting proof aggregator service",);
 
         info!("About to aggregate and submit proof to be verified on chain");
         let res = self.aggregate_and_submit_proofs_on_chain().await;
 
         match res {
             Ok(()) => {
+                config
+                    .update_last_aggregated_block(self.fetcher.get_last_aggregated_block())
+                    .unwrap();
                 info!("Process finished successfully");
             }
             Err(err) => {

aggregation_mode/src/backend/mod.rs

@@ -27,5 +27,5 @@ async fn main() {
     tracing::info!("Config loaded");
 
     let mut proof_aggregator = ProofAggregator::new(&config);
-    proof_aggregator.start().await;
+    proof_aggregator.start(&config).await;
 }

aggregation_mode/src/main.rs

@@ -25,3 +25,4 @@ hex = "0.4.3"
 ciborium = "=0.2.2"
 serde_repr = "0.1.19"
 dialoguer = "0.11.0"
+reqwest = { version = "0.12", features = ["json"] }