feat(eigen-client-extra-features): address PR comments (part 4)

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,

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);

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<Option<Vec<u8>>>;
+
+    fn clone_boxed(&self) -> Box<dyn GetBlobData>;
 }
 
 /// EigenClient is a client for the Eigen DA service.
 #[derive(Debug, Clone)]
-pub struct EigenClient<T: GetBlobData> {
-    pub(crate) client: Arc<RawEigenClient<T>>,
+pub struct EigenClient {
+    pub(crate) client: Arc<RawEigenClient>,
 }
 
-impl<T: GetBlobData> EigenClient<T> {
+impl EigenClient {
     pub async fn new(
         config: EigenConfig,
         secrets: EigenSecrets,
-        get_blob_data: Box<T>,
+        get_blob_data: Box<dyn GetBlobData>,
     ) -> anyhow::Result<Self> {
         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<T: GetBlobData> EigenClient<T> {
 }
 
 #[async_trait]
-impl<T: GetBlobData + 'static> DataAvailabilityClient for EigenClient<T> {
+impl DataAvailabilityClient for EigenClient {
     async fn dispatch_blob(
         &self,
         _: u32, // batch number
@@ -75,6 +77,6 @@ impl<T: GetBlobData + 'static> DataAvailabilityClient for EigenClient<T> {
     }
 
     fn blob_size_limit(&self) -> Option<usize> {
-        Some(RawEigenClient::<T>::blob_size_limit())
+        Some(RawEigenClient::blob_size_limit())
     }
 }

core/node/da_clients/src/eigen/client_tests.rs

@@ -17,7 +17,7 @@ mod tests {
 
     use crate::eigen::{blob_info::BlobInfo, EigenClient, GetBlobData};
 
-    impl<T: GetBlobData> EigenClient<T> {
+    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<T: GetBlobData>(
-        client: &EigenClient<T>,
+    async fn get_blob_info(
+        client: &EigenClient,
         result: &DispatchResponse,
     ) -> anyhow::Result<BlobInfo> {
         let blob_info = (|| async {
@@ -62,6 +62,10 @@ mod tests {
         async fn get_blob_data(&self, _input: &'_ str) -> anyhow::Result<Option<Vec<u8>>> {
             Ok(None)
         }
+
+        fn clone_boxed(&self) -> Box<dyn GetBlobData> {
+            Box::new(self.clone())
+        }
     }
 
     #[ignore = "depends on external RPC"]

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<T: GetBlobData> {
+#[derive(Debug)]
+pub(crate) struct RawEigenClient {
     client: Arc<Mutex<DisperserClient<Channel>>>,
     private_key: SecretKey,
     pub config: EigenConfig,
     verifier: Verifier,
-    get_blob_data: Box<T>,
+    get_blob_data: Box<dyn GetBlobData>,
+}
+
+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<T: GetBlobData> RawEigenClient<T> {
+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<T>,
+        get_blob_data: Box<dyn GetBlobData>,
     ) -> anyhow::Result<Self> {
         let endpoint =
             Endpoint::from_str(config.disperser_rpc.as_str())?.tls_config(ClientTlsConfig::new())?;
@@ -60,8 +73,8 @@ impl<T: GetBlobData> RawEigenClient<T> {
                 .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,

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<String, VerificationError> {
-        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<String, VerificationError> {
+        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<Self, VerificationError> {
         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<u8>) -> Result<Vec<u8>, 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<u8>) -> Result<Vec<u8>, VerificationError> {
+        let output_type = [ParamType::Bytes];
+        let tokens: Vec<Token> = 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

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(),

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<dyn GetBlobData> {
+        Box::new(self.clone())
+    }
 }