feat: basic fetcher flow

Author: MarcosNicolau

aggregation_mode/proof_aggregator/src/aggregators/mod.rs

@@ -3,10 +3,11 @@ pub mod sp1_aggregator;
 
 use std::fmt::Display;
 
+use aligned_sdk::aggregation_layer::AggregationModeProvingSystem;
 use lambdaworks_crypto::merkle_tree::traits::IsMerkleTreeBackend;
 use risc0_aggregator::{Risc0AggregationError, Risc0ProofReceiptAndImageId};
 use sha3::{Digest, Keccak256};
-use sp1_aggregator::{SP1AggregationError, SP1ProofWithPubValuesAndElf};
+use sp1_aggregator::{SP1AggregationError, SP1ProofWithPubValuesAndVk};
 use tracing::info;
 
 #[derive(Clone, Debug)]
@@ -44,6 +45,13 @@ impl ZKVMEngine {
         Some(engine)
     }
 
+    pub fn proving_system_id(&self) -> u16 {
+        match &self {
+            ZKVMEngine::SP1 => AggregationModeProvingSystem::SP1.as_u16(),
+            ZKVMEngine::RISC0 => AggregationModeProvingSystem::RISC0.as_u16(),
+        }
+    }
+
     /// Aggregates a list of [`AlignedProof`]s into a single [`AlignedProof`].
     ///
     /// Returns a tuple containing:
@@ -61,7 +69,7 @@ impl ZKVMEngine {
     ) -> Result<(AlignedProof, [u8; 32]), ProofAggregationError> {
         let res = match self {
             ZKVMEngine::SP1 => {
-                let proofs: Vec<SP1ProofWithPubValuesAndElf> = proofs
+                let proofs: Vec<SP1ProofWithPubValuesAndVk> = proofs
                     .into_iter()
                     // Fetcher already filtered for SP1
                     // We do this for type casting, as to avoid using generics
@@ -80,7 +88,7 @@ impl ZKVMEngine {
                     proofs_per_chunk,
                 );
 
-                let mut agg_proofs: Vec<(SP1ProofWithPubValuesAndElf, Vec<[u8; 32]>)> = vec![];
+                let mut agg_proofs: Vec<(SP1ProofWithPubValuesAndVk, Vec<[u8; 32]>)> = vec![];
                 for (i, chunk) in chunks.enumerate() {
                     let leaves_commitment =
                         chunk.iter().map(|e| e.hash_vk_and_pub_inputs()).collect();
@@ -154,7 +162,7 @@ impl ZKVMEngine {
 }
 
 pub enum AlignedProof {
-    SP1(Box<SP1ProofWithPubValuesAndElf>),
+    SP1(Box<SP1ProofWithPubValuesAndVk>),
     Risc0(Box<Risc0ProofReceiptAndImageId>),
 }
 

aggregation_mode/proof_aggregator/src/aggregators/sp1_aggregator.rs

@@ -25,9 +25,8 @@ static SP1_PROVER_CLIENT: LazyLock<EnvProver> = LazyLock::new(ProverClient::from
 static SP1_PROVER_CLIENT_CPU: LazyLock<CpuProver> =
     LazyLock::new(|| ProverClient::builder().cpu().build());
 
-pub struct SP1ProofWithPubValuesAndElf {
+pub struct SP1ProofWithPubValuesAndVk {
     pub proof_with_pub_values: SP1ProofWithPublicValues,
-    pub elf: Vec<u8>,
     pub vk: SP1VerifyingKey,
 }
 
@@ -37,16 +36,14 @@ pub enum AlignedSP1VerificationError {
     UnsupportedProof,
 }
 
-impl SP1ProofWithPubValuesAndElf {
+impl SP1ProofWithPubValuesAndVk {
     /// Constructs a new instance of the struct by verifying a given SP1 proof with its public values.
     pub fn new(
         proof_with_pub_values: SP1ProofWithPublicValues,
-        elf: Vec<u8>,
+        vk: SP1VerifyingKey,
     ) -> Result<Self, AlignedSP1VerificationError> {
         let client = &*SP1_PROVER_CLIENT_CPU;
 
-        let (_pk, vk) = client.setup(&elf);
-
         // only sp1 compressed proofs are supported for aggregation now
         match proof_with_pub_values.proof {
             sp1_sdk::SP1Proof::Compressed(_) => client
@@ -57,7 +54,6 @@ impl SP1ProofWithPubValuesAndElf {
 
         Ok(Self {
             proof_with_pub_values,
-            elf,
             vk,
         })
     }
@@ -80,8 +76,8 @@ pub enum SP1AggregationError {
 }
 
 pub(crate) fn run_user_proofs_aggregator(
-    proofs: &[SP1ProofWithPubValuesAndElf],
-) -> Result<SP1ProofWithPubValuesAndElf, SP1AggregationError> {
+    proofs: &[SP1ProofWithPubValuesAndVk],
+) -> Result<SP1ProofWithPubValuesAndVk, SP1AggregationError> {
     let mut stdin = SP1Stdin::new();
 
     let mut program_input = sp1_aggregation_program::UserProofsAggregatorInput {
@@ -131,18 +127,17 @@ pub(crate) fn run_user_proofs_aggregator(
         .verify(&proof, &vk)
         .map_err(SP1AggregationError::Verification)?;
 
-    let proof_and_elf = SP1ProofWithPubValuesAndElf {
+    let proof_and_elf = SP1ProofWithPubValuesAndVk {
         proof_with_pub_values: proof,
-        elf: USER_PROOFS_PROGRAM_ELF.to_vec(),
         vk,
     };
 
     Ok(proof_and_elf)
 }
 
 pub(crate) fn run_chunk_aggregator(
-    proofs: &[(SP1ProofWithPubValuesAndElf, Vec<[u8; 32]>)],
-) -> Result<SP1ProofWithPubValuesAndElf, SP1AggregationError> {
+    proofs: &[(SP1ProofWithPubValuesAndVk, Vec<[u8; 32]>)],
+) -> Result<SP1ProofWithPubValuesAndVk, SP1AggregationError> {
     let mut stdin = SP1Stdin::new();
 
     let mut program_input = sp1_aggregation_program::ChunkAggregatorInput {
@@ -204,9 +199,8 @@ pub(crate) fn run_chunk_aggregator(
         .verify(&proof, &vk)
         .map_err(SP1AggregationError::Verification)?;
 
-    let proof_and_elf = SP1ProofWithPubValuesAndElf {
+    let proof_and_elf = SP1ProofWithPubValuesAndVk {
         proof_with_pub_values: proof,
-        elf: CHUNK_PROGRAM_ELF.to_vec(),
         vk,
     };
 

aggregation_mode/proof_aggregator/src/backend/config.rs

@@ -1,28 +1,22 @@
 use serde::{Deserialize, Serialize};
-use std::{fs::File, fs::OpenOptions, io::Read, io::Write};
+use std::{fs::File, io::Read};
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct ECDSAConfig {
     pub private_key_store_path: String,
     pub private_key_store_password: String,
 }
 
-#[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 proofs_per_chunk: u16,
-    pub total_proofs_limit: u16,
+    pub total_proofs_limit: i64,
     pub sp1_chunk_aggregator_vk_hash: String,
     pub risc0_chunk_aggregator_image_id: String,
     pub db_connection_url: String,
@@ -36,32 +30,4 @@ 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/proof_aggregator/src/backend/db.rs

@@ -1,9 +1,5 @@
 use db::types::Task;
-use sqlx::{
-    postgres::PgPoolOptions,
-    types::{BigDecimal, Uuid},
-    Pool, Postgres,
-};
+use sqlx::{postgres::PgPoolOptions, Pool, Postgres};
 
 #[derive(Clone, Debug)]
 pub struct Db {
@@ -29,13 +25,17 @@ impl Db {
 
     pub async fn get_pending_tasks_and_mark_them_as_processed(
         &self,
+        proving_system_id: i64,
         limit: i64,
     ) -> Result<Vec<Task>, DbError> {
-        sqlx::query_as::<_, Task>("SELECT * FROM tasks WHERE status = 'pending' LIMIT $1")
-            .bind(limit)
-            .fetch_all(&self.pool)
-            .await
-            .map_err(|e| DbError::Query(e.to_string()))
+        sqlx::query_as::<_, Task>(
+            "SELECT * FROM tasks WHERE status = 'pending' AND proving_system_id = $1 LIMIT $2",
+        )
+        .bind(proving_system_id)
+        .bind(limit)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| DbError::Query(e.to_string()))
     }
 
     pub async fn mark_tasks_as_pending(&self) {}