feat(aggregation_mode): save last processed block in a file (#1872)

Author: Mechanix97

.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
 

aggregation_mode/src/backend/config.rs

@@ -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/fetcher.rs

@@ -24,8 +24,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 {
@@ -38,31 +37,49 @@ 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) -> Result<Vec<AlignedProof>, ProofsFetcherError> {
-        let from_block = self.get_block_number_to_fetch_from().await?;
+    pub async fn fetch(&mut self) -> 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 {
@@ -119,15 +136,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/mod.rs

@@ -69,14 +69,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/main.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;
 }

config-files/config-proof-aggregator-mock.yaml

@@ -1,13 +1,9 @@
-aligned_service_manager_address: "0x851356ae760d987E095750cCeb3bC6014560891C"
-proof_aggregation_service_address: "0xB0D4afd8879eD9F52b28595d31B441D079B2Ca07"
-eth_rpc_url: "http://localhost:8545"
-eth_ws_url: "ws://localhost:8545"
+eth_rpc_url: http://localhost:8545
+eth_ws_url: ws://localhost:8545
 max_proofs_in_queue: 1000
-# How far in the past should the service go to fetch batch logs
-fetch_logs_from_secs_ago: 86400 # 24hs
-# Anvil start with block time is 7 seconds
-block_time_secs: 7
-
+proof_aggregation_service_address: 0xB0D4afd8879eD9F52b28595d31B441D079B2Ca07
+aligned_service_manager_address: 0x851356ae760d987E095750cCeb3bC6014560891C
+last_aggregated_block_filepath: config-files/proof-aggregator.last_aggregated_block.json
 ecdsa:
-  private_key_store_path: "config-files/anvil.proof-aggregator.ecdsa.key.json"
-  private_key_store_password: ""
+  private_key_store_path: config-files/anvil.proof-aggregator.ecdsa.key.json
+  private_key_store_password: ''

config-files/config-proof-aggregator.yaml

@@ -3,10 +3,7 @@ proof_aggregation_service_address: "0xcbEAF3BDe82155F56486Fb5a1072cb8baAf547cc"
 eth_rpc_url: "http://localhost:8545"
 eth_ws_url: "ws://localhost:8545"
 max_proofs_in_queue: 1000
-# How far in the past should the service go to fetch batch logs
-fetch_logs_from_secs_ago: 86400 # 24hs
-# Anvil start with block time is 7 seconds
-block_time_secs: 7
+last_aggregated_block_filepath: config-files/proof-aggregator.last_aggregated_block.json
 
 ecdsa:
     private_key_store_path: "config-files/anvil.proof-aggregator.ecdsa.key.json"

config-files/proof-aggregator.last_aggregated_block.json

@@ -0,0 +1 @@
+{"last_aggregated_block":0}