feat: add to sdk an option to wait for the verification of the batcher in ethereum make it the default (#583)

Co-authored-by: Mariano A. Nicolini <mariano.nicolini.91@gmail.com>
Co-authored-by: Tatu <65305492+srosati@users.noreply.github.com>

Author: 3 people

batcher/aligned-batcher/src/config/mod.rs

@@ -42,8 +42,6 @@ impl ConfigFromYaml {
 
 #[derive(Debug, Deserialize)]
 pub struct Addresses {
-    #[serde(rename = "alignedLayerServiceManager")]
-    pub aligned_layer_service_manager: String,
     #[serde(rename = "batcherPaymentService")]
     pub batcher_payment_service: String,
 }

batcher/aligned-batcher/src/eth/mod.rs

@@ -3,7 +3,6 @@ use std::sync::Arc;
 
 use ethers::prelude::k256::ecdsa::SigningKey;
 use ethers::prelude::*;
-use stream::EventStream;
 
 use crate::config::ECDSAConfig;
 
@@ -22,45 +21,13 @@ pub struct BatchVerified {
     pub batch_merkle_root: [u8; 32],
 }
 
-pub type AlignedLayerServiceManager =
-    AlignedLayerServiceManagerContract<SignerMiddleware<Provider<Http>, Wallet<SigningKey>>>;
-
-pub type BatchVerifiedEventStream<'s> = EventStream<
-    's,
-    FilterWatcher<'s, Http, Log>,
-    BatchVerifiedFilter,
-    ContractError<SignerMiddleware<Provider<Http>, Wallet<SigningKey>>>,
->;
-
 pub type BatcherPaymentService =
     BatcherPaymentServiceContract<SignerMiddleware<Provider<Http>, Wallet<SigningKey>>>;
 
 pub fn get_provider(eth_rpc_url: String) -> Result<Provider<Http>, anyhow::Error> {
     Provider::<Http>::try_from(eth_rpc_url).map_err(|err| anyhow::anyhow!(err))
 }
 
-pub async fn get_service_manager(
-    provider: Provider<Http>,
-    ecdsa_config: ECDSAConfig,
-    contract_address: String,
-) -> Result<AlignedLayerServiceManager, anyhow::Error> {
-    let chain_id = provider.get_chainid().await?;
-
-    // get private key from keystore
-    let wallet = Wallet::decrypt_keystore(
-        &ecdsa_config.private_key_store_path,
-        &ecdsa_config.private_key_store_password,
-    )?
-    .with_chain_id(chain_id.as_u64());
-
-    let signer = Arc::new(SignerMiddleware::new(provider, wallet));
-
-    let service_manager =
-        AlignedLayerServiceManager::new(H160::from_str(contract_address.as_str())?, signer);
-
-    Ok(service_manager)
-}
-
 pub async fn create_new_task(
     payment_service: &BatcherPaymentService,
     batch_merkle_root: [u8; 32],

batcher/aligned-batcher/src/gnark/mod.rs

@@ -1,4 +1,4 @@
-use aligned_sdk::types::ProvingSystemId;
+use aligned_sdk::core::types::ProvingSystemId;
 
 #[derive(Copy, Clone, Debug)]
 #[repr(C)]

batcher/aligned-batcher/src/lib.rs

@@ -2,19 +2,16 @@ extern crate core;
 
 use dotenv::dotenv;
 
-use std::borrow::Cow;
 use std::env;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::Duration;
 
-use crate::eth::BatchVerifiedEventStream;
-use aligned_sdk::types::{
-    BatchInclusionData, ClientMessage, VerificationCommitmentBatch, VerificationData,
-    VerificationDataCommitment,
+use aligned_sdk::core::types::{
+    BatchInclusionData, ClientMessage, ResponseMessage, VerificationCommitmentBatch,
+    VerificationData, VerificationDataCommitment,
 };
 use aws_sdk_s3::client::Client as S3Client;
-use eth::{BatchVerifiedFilter, BatcherPaymentService};
+use eth::BatcherPaymentService;
 use ethers::prelude::{Middleware, Provider};
 use ethers::providers::Ws;
 use ethers::types::{Address, U256};
@@ -24,16 +21,13 @@ use lambdaworks_crypto::merkle_tree::merkle::MerkleTree;
 use log::{debug, error, info, warn};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{Mutex, RwLock};
-use tokio::time::timeout;
 use tokio_tungstenite::tungstenite::error::ProtocolError;
-use tokio_tungstenite::tungstenite::protocol::{frame::coding::CloseCode, CloseFrame};
 use tokio_tungstenite::tungstenite::{Error, Message};
 use tokio_tungstenite::WebSocketStream;
 use types::batch_queue::BatchQueue;
 use types::errors::BatcherError;
 
 use crate::config::{ConfigFromYaml, ContractDeploymentOutput, NonPayingConfig};
-use crate::eth::AlignedLayerServiceManager;
 
 mod config;
 mod eth;
@@ -49,7 +43,6 @@ pub struct Batcher {
     s3_client: S3Client,
     s3_bucket_name: String,
     eth_ws_provider: Provider<Ws>,
-    service_manager: AlignedLayerServiceManager,
     payment_service: BatcherPaymentService,
     batch_queue: Mutex<BatchQueue>,
     max_block_interval: u64,
@@ -90,14 +83,6 @@ impl Batcher {
             .try_into()
             .unwrap();
 
-        let service_manager = eth::get_service_manager(
-            eth_rpc_provider.clone(),
-            config.ecdsa.clone(),
-            deployment_output.addresses.aligned_layer_service_manager,
-        )
-        .await
-        .expect("Failed to get Aligned service manager contract");
-
         let payment_service = eth::get_batcher_payment_service(
             eth_rpc_provider,
             config.ecdsa,
@@ -115,7 +100,6 @@ impl Batcher {
             s3_client,
             s3_bucket_name,
             eth_ws_provider,
-            service_manager,
             payment_service,
             batch_queue: Mutex::new(BatchQueue::new()),
             max_block_interval: config.batcher.block_interval,
@@ -173,19 +157,19 @@ impl Batcher {
         let (outgoing, incoming) = ws_stream.split();
         let outgoing = Arc::new(RwLock::new(outgoing));
 
-        // Send the protocol version to the client
-        let protocol_version_msg = Message::binary(
-            aligned_sdk::sdk::CURRENT_PROTOCOL_VERSION
-                .to_be_bytes()
-                .to_vec(),
+        let protocol_version_msg = ResponseMessage::ProtocolVersion(
+            aligned_sdk::communication::protocol::EXPECTED_PROTOCOL_VERSION,
         );
 
+        let serialized_protocol_version_msg = serde_json::to_vec(&protocol_version_msg)
+            .expect("Could not serialize protocol version message");
+
         outgoing
             .write()
             .await
-            .send(protocol_version_msg)
+            .send(Message::binary(serialized_protocol_version_msg))
             .await
-            .expect("Failed to send protocol version");
+            .expect("Could not send protocol version message");
 
         match incoming
             .try_filter(|msg| future::ready(msg.is_text()))
@@ -357,7 +341,6 @@ impl Batcher {
         &self,
         block_number: u64,
         finalized_batch: BatchQueue,
-        wait_for_verification: bool,
     ) -> Result<(), BatcherError> {
         let batch_verification_data: Vec<VerificationData> = finalized_batch
             .clone()
@@ -383,12 +366,6 @@ impl Batcher {
             .map(|(_, _, _, addr)| *addr)
             .collect();
 
-        let events = self.service_manager.event::<BatchVerifiedFilter>();
-        let mut stream = events
-            .stream()
-            .await
-            .map_err(|e| BatcherError::BatchVerifiedEventStreamError(e.to_string()))?;
-
         {
             let mut last_uploaded_batch_block = self.last_uploaded_batch_block.lock().await;
             // update last uploaded batch block
@@ -402,24 +379,7 @@ impl Batcher {
         self.submit_batch(&batch_bytes, &batch_merkle_tree.root, submitter_addresses)
             .await;
 
-        if !wait_for_verification {
-            send_batch_inclusion_data_responses(finalized_batch, &batch_merkle_tree).await;
-            return Ok(());
-        }
-
-        // This future is created to be passed to the timeout function, so that if it is not resolved
-        // within the timeout interval an error is raised. If the event is received, responses are sent to
-        // connected clients
-        let await_batch_verified_fut =
-            await_batch_verified_event(&mut stream, &batch_merkle_tree.root);
-        if timeout(Duration::from_secs(60), await_batch_verified_fut)
-            .await
-            .is_err()
-        {
-            send_timeout_close(finalized_batch).await?;
-        } else {
-            send_batch_inclusion_data_responses(finalized_batch, &batch_merkle_tree).await;
-        }
+        send_batch_inclusion_data_responses(finalized_batch, &batch_merkle_tree).await;
 
         Ok(())
     }
@@ -428,8 +388,7 @@ impl Batcher {
     /// finalizes the batch.
     async fn handle_new_block(&self, block_number: u64) -> Result<(), BatcherError> {
         while let Some(finalized_batch) = self.is_batch_ready(block_number).await {
-            self.finalize_batch(block_number, finalized_batch, false)
-                .await?;
+            self.finalize_batch(block_number, finalized_batch).await?;
         }
         Ok(())
     }
@@ -489,26 +448,6 @@ impl Batcher {
         }
     }
 }
-/// Await for the `BatchVerified` event emitted by the Aligned contract and then send responses.
-async fn await_batch_verified_event<'s>(
-    events_stream: &mut BatchVerifiedEventStream<'s>,
-    batch_merkle_root: &[u8; 32],
-) -> Result<(), BatcherError> {
-    while let Some(event_result) = events_stream.next().await {
-        if let Ok(event) = event_result {
-            if &event.batch_merkle_root == batch_merkle_root {
-                info!("Batch operator signatures verified on Ethereum. Sending response to clients...");
-                break;
-            }
-        } else {
-            error!("Error awaiting for batch signature verification event");
-            return Err(BatcherError::BatchVerifiedEventStreamError(
-                event_result.unwrap_err().to_string(),
-            ));
-        }
-    }
-    Ok(())
-}
 
 async fn send_batch_inclusion_data_responses(
     finalized_batch: BatchQueue,
@@ -517,7 +456,9 @@ async fn send_batch_inclusion_data_responses(
     stream::iter(finalized_batch.iter())
         .enumerate()
         .for_each(|(vd_batch_idx, (_, _, ws_sink, _))| async move {
-            let response = BatchInclusionData::new(vd_batch_idx, batch_merkle_tree);
+            let batch_inclusion_data = BatchInclusionData::new(vd_batch_idx, batch_merkle_tree);
+            let response = ResponseMessage::BatchInclusionData(batch_inclusion_data);
+
             let serialized_response =
                 serde_json::to_vec(&response).expect("Could not serialize response");
 
@@ -537,30 +478,3 @@ async fn send_batch_inclusion_data_responses(
         })
         .await;
 }
-
-/// Send a close response to all clients that included data in the batch indicated that a
-/// timeout was exceeded awaiting for the batch verification events
-async fn send_timeout_close(finalized_batch: BatchQueue) -> Result<(), BatcherError> {
-    let timeout_msg = Message::Close(Some(CloseFrame {
-        code: CloseCode::Protocol,
-        reason: Cow::from("Timeout: BatchVerified event not received"),
-    }));
-
-    for (_, _, ws_sink, _) in finalized_batch.iter() {
-        let send_result = ws_sink.write().await.send(timeout_msg.clone()).await;
-        match send_result {
-            // When two or more proofs from the same client are included into a batch,
-            // there will be more than one `ws_sink` corresponding to that client. When one is
-            // closed, the other ones will raise this error. We can just ignore it.
-            Err(Error::Protocol(ProtocolError::SendAfterClosing)) => (),
-            Err(e) => {
-                error!("Error sending timeout response to clients: {}", e);
-                return Err(e.into());
-            }
-            Ok(_) => (),
-        }
-
-        info!("Timeout close response sent");
-    }
-    Ok(())
-}

batcher/aligned-batcher/src/types/batch_queue.rs

@@ -5,7 +5,7 @@ use futures_util::stream::SplitSink;
 use tokio::{net::TcpStream, sync::RwLock};
 use tokio_tungstenite::{tungstenite::Message, WebSocketStream};
 
-use aligned_sdk::types::{VerificationData, VerificationDataCommitment};
+use aligned_sdk::core::types::{VerificationData, VerificationDataCommitment};
 
 pub(crate) type BatchQueueEntry = (
     VerificationData,

batcher/aligned-batcher/src/zk_utils/mod.rs

@@ -1,12 +1,10 @@
-use log::{debug, warn};
-
-use aligned_sdk::types::{ProvingSystemId, VerificationData};
-
 use crate::gnark::verify_gnark;
 use crate::halo2::ipa::verify_halo2_ipa;
 use crate::halo2::kzg::verify_halo2_kzg;
 use crate::risc_zero::verify_risc_zero_proof;
 use crate::sp1::verify_sp1_proof;
+use aligned_sdk::core::types::{ProvingSystemId, VerificationData};
+use log::{debug, warn};
 
 pub(crate) fn verify(verification_data: &VerificationData) -> bool {
     match verification_data.proving_system {