feat: remove queue and returns proofs from fetcher

Author: MarcosNicolau

aggregation-mode/src/backend/config.rs

@@ -17,7 +17,6 @@ pub struct Config {
     pub proof_aggregation_service_address: String,
     pub aligned_service_manager_address: String,
     pub ecdsa: ECDSAConfig,
-    pub bucket_name: String,
 }
 
 impl Config {

aggregation-mode/src/backend/fetcher.rs

@@ -2,7 +2,6 @@ use std::str::FromStr;
 
 use super::{
     config::Config,
-    queue::ProofsQueue,
     types::{AlignedLayerServiceManager, AlignedLayerServiceManagerContract},
 };
 use crate::{
@@ -13,11 +12,11 @@ use aligned_sdk::core::types::ProvingSystemId;
 use alloy::{primitives::Address, providers::ProviderBuilder};
 use tracing::{error, info};
 
-/// This services is in charge of:
-/// 1. Listens to aligned new batch task
-/// 2. Downloads proofs from S3 bucket
-/// 3. Filter supported proofs to be aggregated
-/// 4. Push the proofs to the queue
+#[derive(Debug)]
+pub enum ProofsFetcherError {
+    QueryingLogs,
+}
+
 pub struct ProofsFetcher {
     aligned_service_manager: AlignedLayerServiceManagerContract,
 }
@@ -37,8 +36,7 @@ impl ProofsFetcher {
         }
     }
 
-    // TODO: remove panic and return proofs instead of taking queue
-    pub async fn fetch(&self, queue: &mut ProofsQueue) {
+    pub async fn fetch(&self) -> Result<Vec<Proof>, ProofsFetcherError> {
         info!("Fetching proofs from batch logs");
         // Subscribe to NewBatch event from AlignedServiceManager
         let logs = self
@@ -47,10 +45,12 @@ impl ProofsFetcher {
             .from_block(0)
             .query()
             .await
-            .expect("to get logs");
+            .map_err(|_| ProofsFetcherError::QueryingLogs)?;
 
         info!("Logs collected {}", logs.len());
 
+        let mut proofs = vec![];
+
         for (batch, _) in logs {
             info!(
                 "New batch submitted, about to process. Batch merkle root {}...",
@@ -69,7 +69,7 @@ impl ProofsFetcher {
             info!("Data downloaded from S3, number of proofs {}", data.len());
 
             // Filter SP1 compressed proofs to and push to queue to be aggregated
-            let proofs: Vec<Proof> = data
+            let proofs_to_add: Vec<Proof> = data
                 .into_iter()
                 .filter_map(|p| match p.proving_system {
                     ProvingSystemId::SP1 => {
@@ -86,15 +86,16 @@ impl ProofsFetcher {
             info!("SP1 proofs filtered, total proofs to add {}", proofs.len());
 
             // try to add them to the queue
-            for proof in proofs {
-                match queue.add_proof(proof) {
-                    Ok(_) => info!(
-                        "New proof added to queue, current length {}",
-                        queue.proofs().len()
-                    ),
-                    Err(e) => error!("Could not add proof, reason: {:?}", e),
+            for proof in proofs_to_add {
+                if let Err(err) = proof.verify() {
+                    error!("Could not add proof, verification failed: {:?}", err);
+                    continue;
                 };
+
+                proofs.push(proof);
             }
         }
+
+        Ok(proofs)
     }
 }

aggregation-mode/src/backend/mod.rs

@@ -1,7 +1,6 @@
 pub mod config;
 pub mod fetcher;
 mod merkle_tree;
-pub mod queue;
 mod s3;
 mod types;
 
@@ -18,28 +17,27 @@ use alloy::{
     signers::local::LocalSigner,
 };
 use config::Config;
-use fetcher::ProofsFetcher;
+use fetcher::{ProofsFetcher, ProofsFetcherError};
 use merkle_tree::compute_proofs_merkle_root;
-use queue::ProofsQueue;
 use sp1_sdk::HashableKey;
 use std::{str::FromStr, time::Duration};
 use tracing::{error, info, warn};
 use types::{AlignedProofAggregationService, AlignedProofAggregationServiceContract};
 
 #[derive(Debug)]
-enum AggregatedProofSubmissionError {
+pub enum AggregatedProofSubmissionError {
     Aggregation(ProofAggregationError),
     SendBlobTransaction,
     SendVerifyAggregatedProofTransaction(alloy::contract::Error),
     ReceiptError(PendingTransactionError),
+    FetchingProofs(ProofsFetcherError),
 }
 
 pub struct ProofAggregator {
     engine: ZKVMEngine,
     submit_proof_every_secs: u64,
     proof_aggregation_service: AlignedProofAggregationServiceContract,
     fetcher: ProofsFetcher,
-    queue: ProofsQueue,
 }
 
 impl ProofAggregator {
@@ -63,7 +61,6 @@ impl ProofAggregator {
             engine: ZKVMEngine::SP1,
             submit_proof_every_secs: config.submit_proofs_every_secs,
             proof_aggregation_service,
-            queue: ProofsQueue::new(config.max_proofs_in_queue),
             fetcher,
         }
     }
@@ -99,8 +96,11 @@ impl ProofAggregator {
     async fn aggregate_and_submit_proofs_on_chain(
         &mut self,
     ) -> Result<(), AggregatedProofSubmissionError> {
-        self.fetcher.fetch(&mut self.queue).await;
-        let proofs = self.queue.clear();
+        let proofs = self
+            .fetcher
+            .fetch()
+            .await
+            .map_err(AggregatedProofSubmissionError::FetchingProofs)?;
 
         if proofs.len() == 0 {
             warn!("No proofs in queue, skipping iteration...");
@@ -130,14 +130,14 @@ impl ProofAggregator {
 
         info!("Sending blob transaction...");
         let blob_tx_hash = self.send_blob_transaction(leaves).await?;
-        info!("Blob transaction sen, hash: {:?}", blob_tx_hash);
+        info!("Blob transaction sent, hash: {:?}", blob_tx_hash);
 
         info!("Sending proof to ProofAggregationService contract...");
         let receipt = self
             .send_proof_to_verify_on_chain(&blob_tx_hash, output.proof)
             .await?;
         info!(
-            "Proof sent anv verified, tx hash {:?}",
+            "Proof sent and verified, tx hash {:?}",
             receipt.transaction_hash
         );
 
@@ -175,7 +175,7 @@ impl ProofAggregator {
     // TODO
     async fn send_blob_transaction(
         &self,
-        leaves: Vec<[u8; 32]>,
+        _leaves: Vec<[u8; 32]>,
     ) -> Result<[u8; 32], AggregatedProofSubmissionError> {
         Ok([0u8; 32])
     }