Merge branch 'refs/heads/staging' into 1864-infraaggregation_mode-add-server-setup-and-config-files

Author: JuArce

.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

@@ -143,14 +143,10 @@ lint_contracts:
 	@cd contracts && npm run lint:sol
 
 anvil_start:
-	@echo "Starting Anvil..."
-	anvil --load-state contracts/scripts/anvil/state/alignedlayer-deployed-anvil-state.json
-
-anvil_start_with_block_time:
 	@echo "Starting Anvil..."
 	anvil --load-state contracts/scripts/anvil/state/alignedlayer-deployed-anvil-state.json --block-time 7
 
-anvil_start_with_block_time_with_more_prefunded_accounts:
+anvil_start_with_more_prefunded_accounts:
 	@echo "Starting Anvil..."
 	anvil --load-state contracts/scripts/anvil/state/alignedlayer-deployed-anvil-state.json --block-time 7 -a 2000
 
@@ -1171,7 +1167,7 @@ setup_local_aligned_all:
 	tmux new-session -d -s aligned_layer
 
 	tmux new-window -t aligned_layer -n anvil
-	tmux send-keys -t aligned_layer 'make anvil_start_with_block_time' C-m
+	tmux send-keys -t aligned_layer 'make anvil_start' C-m
 
 	tmux new-window -t aligned_layer -n aggregator
 	tmux send-keys -t aligned_layer:aggregator 'make aggregator_start' C-m

aggregation_mode/aggregation_programs/sp1/src/lib.rs

@@ -2,37 +2,37 @@ use serde::{Deserialize, Serialize};
 use sha3::{Digest, Keccak256};
 
 #[derive(Serialize, Deserialize)]
-pub struct SP1ProofInput {
+pub struct SP1VkAndPubInputs {
     pub vk: [u32; 8],
     pub public_inputs: Vec<u8>,
 }
 
-impl SP1ProofInput {
+impl SP1VkAndPubInputs {
     pub fn hash(&self) -> [u8; 32] {
         let mut hasher = Keccak256::new();
         for &word in &self.vk {
-            hasher.update(word.to_le_bytes());
+            hasher.update(word.to_be_bytes());
         }
         hasher.update(&self.public_inputs);
         hasher.finalize().into()
     }
 }
 
 #[derive(Serialize, Deserialize)]
-pub enum ProofInput {
-    SP1Compressed(SP1ProofInput),
+pub enum ProofVkAndPubInputs {
+    SP1Compressed(SP1VkAndPubInputs),
 }
 
-impl ProofInput {
+impl ProofVkAndPubInputs {
     pub fn hash(&self) -> [u8; 32] {
         match self {
-            ProofInput::SP1Compressed(proof) => proof.hash(),
+            ProofVkAndPubInputs::SP1Compressed(proof_data) => proof_data.hash(),
         }
     }
 }
 
 #[derive(Serialize, Deserialize)]
 pub struct Input {
-    pub proofs: Vec<ProofInput>,
+    pub proofs_vk_and_pub_inputs: Vec<ProofVkAndPubInputs>,
     pub merkle_root: [u8; 32],
 }

aggregation_mode/aggregation_programs/sp1/src/main.rs

@@ -3,7 +3,7 @@ sp1_zkvm::entrypoint!(main);
 
 use sha2::{Digest, Sha256};
 use sha3::Keccak256;
-use sp1_aggregation_program::{Input, ProofInput};
+use sp1_aggregation_program::{Input, ProofVkAndPubInputs};
 
 fn combine_hashes(hash_a: &[u8; 32], hash_b: &[u8; 32]) -> [u8; 32] {
     let mut hasher = Keccak256::new();
@@ -13,7 +13,7 @@ fn combine_hashes(hash_a: &[u8; 32], hash_b: &[u8; 32]) -> [u8; 32] {
 }
 
 /// Computes the merkle root for the given proofs using the vk
-fn compute_merkle_root(proofs: &[ProofInput]) -> [u8; 32] {
+fn compute_merkle_root(proofs: &[ProofVkAndPubInputs]) -> [u8; 32] {
     let mut leaves: Vec<[u8; 32]> = proofs
         .chunks(2)
         .map(|chunk| match chunk {
@@ -41,9 +41,9 @@ pub fn main() {
     let input = sp1_zkvm::io::read::<Input>();
 
     // Verify the proofs.
-    for proof in input.proofs.iter() {
+    for proof in input.proofs_vk_and_pub_inputs.iter() {
         match proof {
-            ProofInput::SP1Compressed(proof) => {
+            ProofVkAndPubInputs::SP1Compressed(proof) => {
                 let vkey = proof.vk;
                 let public_values = &proof.public_inputs;
                 let public_values_digest = Sha256::digest(public_values);
@@ -52,7 +52,7 @@ pub fn main() {
         }
     }
 
-    let merkle_root = compute_merkle_root(&input.proofs);
+    let merkle_root = compute_merkle_root(&input.proofs_vk_and_pub_inputs);
 
     assert_eq!(merkle_root, input.merkle_root);
 

aggregation_mode/src/aggregators/sp1_aggregator.rs

@@ -1,14 +1,19 @@
+use std::sync::LazyLock;
+
 use alloy::primitives::Keccak256;
-use sp1_aggregation_program::{ProofInput, SP1ProofInput};
+use sp1_aggregation_program::{ProofVkAndPubInputs, SP1VkAndPubInputs};
 use sp1_sdk::{
-    HashableKey, Prover, ProverClient, SP1ProofWithPublicValues, SP1Stdin, SP1VerifyingKey,
+    EnvProver, HashableKey, Prover, ProverClient, SP1ProofWithPublicValues, SP1Stdin,
+    SP1VerifyingKey,
 };
 
 use super::lib::{AggregatedProof, ProgramOutput, ProofAggregationError};
 
 const PROGRAM_ELF: &[u8] =
     include_bytes!("../../aggregation_programs/sp1/elf/sp1_aggregator_program");
 
+static SP1_PROVER_CLIENT: LazyLock<EnvProver> = LazyLock::new(ProverClient::from_env);
+
 pub struct SP1ProofWithPubValuesAndElf {
     pub proof_with_pub_values: SP1ProofWithPublicValues,
     pub elf: Vec<u8>,
@@ -17,9 +22,8 @@ pub struct SP1ProofWithPubValuesAndElf {
 impl SP1ProofWithPubValuesAndElf {
     pub fn hash_vk_and_pub_inputs(&self) -> [u8; 32] {
         let mut hasher = Keccak256::new();
-        for &word in &self.vk().hash_u32() {
-            hasher.update(word.to_le_bytes());
-        }
+        let vk_bytes = &self.vk().hash_bytes();
+        hasher.update(vk_bytes);
         hasher.update(self.proof_with_pub_values.public_values.as_slice());
         hasher.finalize().into()
     }
@@ -40,15 +44,15 @@ pub(crate) fn aggregate_proofs(
     let mut stdin = SP1Stdin::new();
 
     let mut program_input = sp1_aggregation_program::Input {
-        proofs: vec![],
+        proofs_vk_and_pub_inputs: vec![],
         merkle_root: input.merkle_root,
     };
 
     // write vk + public inputs
     for proof in input.proofs.iter() {
         program_input
-            .proofs
-            .push(ProofInput::SP1Compressed(SP1ProofInput {
+            .proofs_vk_and_pub_inputs
+            .push(ProofVkAndPubInputs::SP1Compressed(SP1VkAndPubInputs {
                 public_inputs: proof.proof_with_pub_values.public_values.to_vec(),
                 vk: proof.vk().hash_u32(),
             }));
@@ -66,7 +70,7 @@ pub(crate) fn aggregate_proofs(
     }
 
     #[cfg(feature = "prove")]
-    let client = ProverClient::from_env();
+    let client = &*SP1_PROVER_CLIENT;
     // If not in prove mode, create a mock proof via mock client
     #[cfg(not(feature = "prove"))]
     let client = ProverClient::builder().mock().build();
@@ -102,7 +106,7 @@ pub enum AlignedSP1VerificationError {
 pub(crate) fn verify(
     sp1_proof_with_pub_values_and_elf: &SP1ProofWithPubValuesAndElf,
 ) -> Result<(), AlignedSP1VerificationError> {
-    let client = ProverClient::from_env();
+    let client = &*SP1_PROVER_CLIENT;
 
     let (_pk, vk) = client.setup(&sp1_proof_with_pub_values_and_elf.elf);
 
@@ -122,7 +126,7 @@ pub(crate) fn verify(
 }
 
 pub fn vk_from_elf(elf: &[u8]) -> SP1VerifyingKey {
-    let prover = ProverClient::builder().cpu().build();
+    let prover = &*SP1_PROVER_CLIENT;
     let (_, vk) = prover.setup(elf);
     vk
 }

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

@@ -17,15 +17,14 @@ use tracing::{error, info};
 
 #[derive(Debug)]
 pub enum ProofsFetcherError {
-    QueryingLogs,
-    BlockNumber,
+    GetLogs(String),
+    GetBlockNumber(String),
 }
 
 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(|_| ProofsFetcherError::QueryingLogs)?;
+            .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(|_| ProofsFetcherError::BlockNumber)?;
-
-        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) => {