diff --git a/core/lib/config/src/configs/da_client/eigen.rs b/core/lib/config/src/configs/da_client/eigen.rs index cd036e407427..b7723e2271a6 100644 --- a/core/lib/config/src/configs/da_client/eigen.rs +++ b/core/lib/config/src/configs/da_client/eigen.rs @@ -1,7 +1,7 @@ use serde::Deserialize; use zksync_basic_types::secrets::PrivateKey; /// Configuration for the EigenDA remote disperser client. -#[derive(Clone, Debug, PartialEq, Deserialize, Default)] +#[derive(Clone, Debug, PartialEq, Deserialize)] pub struct EigenConfig { /// URL of the Disperser RPC server pub disperser_rpc: String, @@ -24,6 +24,22 @@ pub struct EigenConfig { pub chain_id: u64, } +impl Default for EigenConfig { + fn default() -> Self { + Self { + disperser_rpc: "https://disperser-holesky.eigenda.xyz:443".to_string(), + settlement_layer_confirmation_depth: 0, + eigenda_eth_rpc: Some("https://ethereum-holesky-rpc.publicnode.com".to_string()), + eigenda_svc_manager_address: "0xD4A7E1Bd8015057293f0D0A557088c286942e84b".to_string(), + wait_for_finalization: false, + authenticated: false, + g1_url: "https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g1.point".to_string(), + g2_url: "https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g2.point.powerOf2".to_string(), + chain_id: 19000, + } + } +} + #[derive(Clone, Debug, PartialEq)] pub struct EigenSecrets { pub private_key: PrivateKey, diff --git a/core/lib/dal/migrations/20241812144402_create_index_data_availability.sql b/core/lib/dal/migrations/20241812144402_create_index_data_availability.sql new file mode 100644 index 000000000000..938d2e09de44 --- /dev/null +++ b/core/lib/dal/migrations/20241812144402_create_index_data_availability.sql @@ -0,0 +1 @@ +CREATE INDEX idx_blob_id_l1_batch_number ON data_availability (blob_id, l1_batch_number); diff --git a/core/node/da_clients/src/eigen/client.rs b/core/node/da_clients/src/eigen/client.rs index ef4baa1c2ad0..5baee9475e92 100644 --- a/core/node/da_clients/src/eigen/client.rs +++ b/core/node/da_clients/src/eigen/client.rs @@ -13,21 +13,23 @@ use super::sdk::RawEigenClient; use crate::utils::to_retriable_da_error; #[async_trait] -pub trait GetBlobData: Clone + std::fmt::Debug + Send + Sync { +pub trait GetBlobData: std::fmt::Debug + Send + Sync { async fn get_blob_data(&self, input: &str) -> anyhow::Result>>; + + fn clone_boxed(&self) -> Box; } /// EigenClient is a client for the Eigen DA service. #[derive(Debug, Clone)] -pub struct EigenClient { - pub(crate) client: Arc>, +pub struct EigenClient { + pub(crate) client: Arc, } -impl EigenClient { +impl EigenClient { pub async fn new( config: EigenConfig, secrets: EigenSecrets, - get_blob_data: Box, + get_blob_data: Box, ) -> anyhow::Result { let private_key = SecretKey::from_str(secrets.private_key.0.expose_secret().as_str()) .map_err(|e| anyhow::anyhow!("Failed to parse private key: {}", e))?; @@ -40,7 +42,7 @@ impl EigenClient { } #[async_trait] -impl DataAvailabilityClient for EigenClient { +impl DataAvailabilityClient for EigenClient { async fn dispatch_blob( &self, _: u32, // batch number @@ -75,6 +77,6 @@ impl DataAvailabilityClient for EigenClient { } fn blob_size_limit(&self) -> Option { - Some(RawEigenClient::::blob_size_limit()) + Some(RawEigenClient::blob_size_limit()) } } diff --git a/core/node/da_clients/src/eigen/client_tests.rs b/core/node/da_clients/src/eigen/client_tests.rs index 99cd81cf2797..4fbbd5c36761 100644 --- a/core/node/da_clients/src/eigen/client_tests.rs +++ b/core/node/da_clients/src/eigen/client_tests.rs @@ -17,7 +17,7 @@ mod tests { use crate::eigen::{blob_info::BlobInfo, EigenClient, GetBlobData}; - impl EigenClient { + impl EigenClient { pub async fn get_blob_data( &self, blob_id: BlobInfo, @@ -32,8 +32,8 @@ mod tests { const STATUS_QUERY_TIMEOUT: u64 = 1800000; // 30 minutes const STATUS_QUERY_INTERVAL: u64 = 5; // 5 ms - async fn get_blob_info( - client: &EigenClient, + async fn get_blob_info( + client: &EigenClient, result: &DispatchResponse, ) -> anyhow::Result { let blob_info = (|| async { @@ -62,6 +62,10 @@ mod tests { async fn get_blob_data(&self, _input: &'_ str) -> anyhow::Result>> { Ok(None) } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } } #[ignore = "depends on external RPC"] diff --git a/core/node/da_clients/src/eigen/sdk.rs b/core/node/da_clients/src/eigen/sdk.rs index c1e52e42ffc9..3a3b1202690c 100644 --- a/core/node/da_clients/src/eigen/sdk.rs +++ b/core/node/da_clients/src/eigen/sdk.rs @@ -7,6 +7,7 @@ use tonic::{ transport::{Channel, ClientTlsConfig, Endpoint}, Streaming, }; +use url::Url; use zksync_config::EigenConfig; use zksync_da_client::types::DAError; use zksync_eth_client::clients::PKSigningClient; @@ -30,24 +31,36 @@ use crate::eigen::{ verifier::VerificationError, }; -#[derive(Debug, Clone)] -pub(crate) struct RawEigenClient { +#[derive(Debug)] +pub(crate) struct RawEigenClient { client: Arc>>, private_key: SecretKey, pub config: EigenConfig, verifier: Verifier, - get_blob_data: Box, + get_blob_data: Box, +} + +impl Clone for RawEigenClient { + fn clone(&self) -> Self { + Self { + client: self.client.clone(), + private_key: self.private_key, + config: self.config.clone(), + verifier: self.verifier.clone(), + get_blob_data: self.get_blob_data.clone_boxed(), + } + } } pub(crate) const DATA_CHUNK_SIZE: usize = 32; -impl RawEigenClient { +impl RawEigenClient { const BLOB_SIZE_LIMIT: usize = 1024 * 1024 * 2; // 2 MB pub async fn new( private_key: SecretKey, config: EigenConfig, - get_blob_data: Box, + get_blob_data: Box, ) -> anyhow::Result { let endpoint = Endpoint::from_str(config.disperser_rpc.as_str())?.tls_config(ClientTlsConfig::new())?; @@ -60,8 +73,8 @@ impl RawEigenClient { .ok_or(anyhow::anyhow!("EigenDA ETH RPC not set"))?, svc_manager_addr: Address::from_str(&config.eigenda_svc_manager_address)?, max_blob_size: Self::BLOB_SIZE_LIMIT as u32, - g1_url: config.g1_url.clone(), - g2_url: config.g2_url.clone(), + g1_url: Url::parse(&config.g1_url)?, + g2_url: Url::parse(&config.g2_url)?, settlement_layer_confirmation_depth: config.settlement_layer_confirmation_depth, private_key: hex::encode(private_key.secret_bytes()), chain_id: config.chain_id, diff --git a/core/node/da_clients/src/eigen/verifier.rs b/core/node/da_clients/src/eigen/verifier.rs index ea9e25b118c9..b6d196df47bc 100644 --- a/core/node/da_clients/src/eigen/verifier.rs +++ b/core/node/da_clients/src/eigen/verifier.rs @@ -1,8 +1,9 @@ -use std::{collections::HashMap, fs::File, io::copy, path::Path}; +use std::{collections::HashMap, path::Path}; use ark_bn254::{Fq, G1Affine}; use ethabi::{encode, ParamType, Token}; use rust_kzg_bn254::{blob::Blob, kzg::Kzg, polynomial::PolynomialFormat}; +use tokio::{fs::File, io::AsyncWriteExt}; use url::Url; use zksync_basic_types::web3::CallRequest; use zksync_eth_client::{clients::PKSigningClient, EnrichedClientResult}; @@ -69,8 +70,8 @@ pub struct VerifierConfig { pub rpc_url: String, pub svc_manager_addr: Address, pub max_blob_size: u32, - pub g1_url: String, - pub g2_url: String, + pub g1_url: Url, + pub g2_url: Url, pub settlement_layer_confirmation_depth: u32, pub private_key: String, pub chain_id: u64, @@ -103,8 +104,7 @@ impl Verifier { pub const G2POINT: &'static str = "g2.point.powerOf2"; pub const POINT_SIZE: u32 = 32; - async fn save_point(url: String, point: String) -> Result<(), VerificationError> { - let url = Url::parse(&url).map_err(|_| VerificationError::LinkError)?; + async fn save_point(url: Url, point: String) -> Result<(), VerificationError> { let response = reqwest::get(url) .await .map_err(|_| VerificationError::LinkError)?; @@ -113,17 +113,19 @@ impl Verifier { } let path = format!("./{}", point); let path = Path::new(&path); - let mut file = File::create(path).map_err(|_| VerificationError::LinkError)?; + let mut file = File::create(path).await.map_err(|_| VerificationError::LinkError)?; let content = response .bytes() .await .map_err(|_| VerificationError::LinkError)?; - copy(&mut content.as_ref(), &mut file).map_err(|_| VerificationError::LinkError)?; + file.write_all(&content) + .await + .map_err(|_| VerificationError::LinkError)?; Ok(()) } - async fn save_points(url_g1: String, url_g2: String) -> Result { - Self::save_point(url_g1.clone(), Self::G1POINT.to_string()).await?; - Self::save_point(url_g2.clone(), Self::G2POINT.to_string()).await?; + async fn save_points(url_g1: Url, url_g2: Url) -> Result { + Self::save_point(url_g1, Self::G1POINT.to_string()).await?; + Self::save_point(url_g2, Self::G2POINT.to_string()).await?; Ok(".".to_string()) } @@ -133,15 +135,20 @@ impl Verifier { ) -> Result { let srs_points_to_load = cfg.max_blob_size / Self::POINT_SIZE; let path = Self::save_points(cfg.clone().g1_url, cfg.clone().g2_url).await?; - let kzg = Kzg::setup( - &format!("{}/{}", path, Self::G1POINT), - "", - &format!("{}/{}", path, Self::G2POINT), - Self::SRSORDER, - srs_points_to_load, - "".to_string(), - ); - let kzg = kzg.map_err(|e| { + let kzg_handle = tokio::task::spawn_blocking(move || { + Kzg::setup( + &format!("{}/{}", path, Self::G1POINT), + "", + &format!("{}/{}", path, Self::G2POINT), + Self::SRSORDER, + srs_points_to_load, + "".to_string(), + ) + }); + let kzg = kzg_handle.await.map_err(|e| { + tracing::error!("Failed to setup KZG: {:?}", e); + VerificationError::KzgError + })?.map_err(|e| { tracing::error!("Failed to setup KZG: {:?}", e); VerificationError::KzgError })?; @@ -295,7 +302,10 @@ impl Verifier { .map_err(|_| VerificationError::ServiceManagerError)? .as_u64(); - let depth = self.cfg.settlement_layer_confirmation_depth.saturating_sub(1); + let depth = self + .cfg + .settlement_layer_confirmation_depth + .saturating_sub(1); let block_to_return = latest.saturating_sub(depth as u64); Ok(block_to_return) } @@ -361,39 +371,17 @@ impl Verifier { Ok(()) } - fn decode_bytes(&self, encoded: Vec) -> Result, String> { - // Ensure the input has at least 64 bytes (offset + length) - if encoded.len() < 64 { - return Err("Encoded data is too short".to_string()); - } - - // Read the offset (first 32 bytes) - let offset = usize::from_be_bytes( - encoded[24..32] - .try_into() - .map_err(|_| "Offset is too large")?, - ); - - // Check if offset is valid - if offset + 32 > encoded.len() { - return Err("Offset points outside the encoded data".to_string()); - } - - // Read the length (32 bytes at the offset position) - let length = usize::from_be_bytes( - encoded[offset + 24..offset + 32] - .try_into() - .map_err(|_| "Length is too large")?, - ); - - // Check if the length is valid - if offset + 32 + length > encoded.len() { - return Err("Length extends beyond the encoded data".to_string()); + fn decode_bytes(&self, encoded: Vec) -> Result, VerificationError> { + let output_type = [ParamType::Bytes]; + let tokens: Vec = ethabi::decode(&output_type, &encoded) + .map_err(|_| VerificationError::ServiceManagerError)?; + let token = tokens + .first() + .ok_or(VerificationError::ServiceManagerError)?; + match token { + Token::Bytes(data) => Ok(data.to_vec()), + _ => Err(VerificationError::ServiceManagerError), } - - // Extract the bytes data - let data = encoded[offset + 32..offset + 32 + length].to_vec(); - Ok(data) } async fn get_quorum_adversary_threshold( @@ -416,9 +404,7 @@ impl Verifier { .await .map_err(|_| VerificationError::ServiceManagerError)?; - let percentages = self - .decode_bytes(res.0.to_vec()) - .map_err(|_| VerificationError::ServiceManagerError)?; + let percentages = self.decode_bytes(res.0.to_vec())?; if percentages.len() > quorum_number as usize { return Ok(percentages[quorum_number as usize]); @@ -443,7 +429,6 @@ impl Verifier { .map_err(|_| VerificationError::ServiceManagerError)?; self.decode_bytes(res.0.to_vec()) - .map_err(|_| VerificationError::ServiceManagerError) } /// Verifies that the certificate's blob quorum params are correct diff --git a/core/node/da_clients/src/eigen/verifier_tests.rs b/core/node/da_clients/src/eigen/verifier_tests.rs index 1d25aa082c8c..35c78dd8d9d3 100644 --- a/core/node/da_clients/src/eigen/verifier_tests.rs +++ b/core/node/da_clients/src/eigen/verifier_tests.rs @@ -2,6 +2,7 @@ mod test { use std::{collections::HashMap, str::FromStr}; + use url::Url; use zksync_eth_client::{clients::PKSigningClient, EnrichedClientResult}; use zksync_types::{ url::SensitiveUrl, @@ -23,8 +24,8 @@ mod test { rpc_url: "https://ethereum-holesky-rpc.publicnode.com".to_string(), svc_manager_addr: Address::from_str("0xD4A7E1Bd8015057293f0D0A557088c286942e84b").unwrap(), max_blob_size: 2 * 1024 * 1024, - g1_url: "https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g1.point".to_string(), - g2_url: "https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g2.point.powerOf2".to_string(), + g1_url: Url::parse("https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g1.point").unwrap(), + g2_url: Url::parse("https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g2.point.powerOf2").unwrap(), settlement_layer_confirmation_depth: 0, private_key: "0xd08aa7ae1bb5ddd46c3c2d8cdb5894ab9f54dec467233686ca42629e826ac4c6" .to_string(), diff --git a/core/node/node_framework/src/implementations/layers/da_clients/eigen.rs b/core/node/node_framework/src/implementations/layers/da_clients/eigen.rs index ee5755d85c14..3eb9675c03be 100644 --- a/core/node/node_framework/src/implementations/layers/da_clients/eigen.rs +++ b/core/node/node_framework/src/implementations/layers/da_clients/eigen.rs @@ -74,4 +74,8 @@ impl GetBlobData for GetBlobFromDB { .await?; Ok(batch.map(|b| b.pubdata)) } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } }