Merge branch 'eigen-client-extra-features' into eigen-client-m0-imple…

…mentation-merge-extra-features

core/lib/config/src/configs/da_client/eigen.rs

@@ -1,13 +1,13 @@
 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,
     /// Block height needed to reach in order to consider the blob finalized
     /// a value less or equal to 0 means that the disperser will not wait for finalization
-    pub settlement_layer_confirmation_depth: i32,
+    pub settlement_layer_confirmation_depth: u32,
     /// URL of the Ethereum RPC server
     pub eigenda_eth_rpc: Option<String>,
     /// Address of the service manager contract
@@ -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/lib/protobuf_config/src/proto/config/da_client.proto

@@ -38,7 +38,7 @@ message CelestiaConfig {
 
 message EigenConfig {
   optional string disperser_rpc = 3;
-  optional int32 settlement_layer_confirmation_depth = 4;
+  optional uint32 settlement_layer_confirmation_depth = 4;
   optional string eigenda_eth_rpc = 5;
   optional string eigenda_svc_manager_address = 6;
   optional bool wait_for_finalization = 7;

core/node/da_clients/Cargo.toml

@@ -62,7 +62,9 @@ ethabi.workspace = true
 rust-kzg-bn254.workspace = true
 ark-bn254.workspace = true
 num-bigint.workspace = true
-serial_test.workspace = true
 zksync_web3_decl.workspace = true
 zksync_eth_client.workspace = true
 url.workspace = true
+
+[dev-dependencies]
+serial_test.workspace = true

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

@@ -12,13 +12,13 @@ use super::{
 
 #[derive(Debug)]
 pub enum ConversionError {
-    NotPresentError,
+    NotPresent,
 }
 
 impl fmt::Display for ConversionError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            ConversionError::NotPresentError => write!(f, "Failed to convert BlobInfo"),
+            ConversionError::NotPresent => write!(f, "Failed to convert BlobInfo"),
         }
     }
 }
@@ -29,18 +29,6 @@ pub struct G1Commitment {
     pub y: Vec<u8>,
 }
 
-impl G1Commitment {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        bytes.extend(&self.x.len().to_be_bytes());
-        bytes.extend(&self.x);
-        bytes.extend(&self.y.len().to_be_bytes());
-        bytes.extend(&self.y);
-
-        bytes
-    }
-}
-
 impl From<DisperserG1Commitment> for G1Commitment {
     fn from(value: DisperserG1Commitment) -> Self {
         Self {
@@ -58,18 +46,6 @@ pub struct BlobQuorumParam {
     pub chunk_length: u32,
 }
 
-impl BlobQuorumParam {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        bytes.extend(&self.quorum_number.to_be_bytes());
-        bytes.extend(&self.adversary_threshold_percentage.to_be_bytes());
-        bytes.extend(&self.confirmation_threshold_percentage.to_be_bytes());
-        bytes.extend(&self.chunk_length.to_be_bytes());
-
-        bytes
-    }
-}
-
 impl From<DisperserBlobQuorumParam> for BlobQuorumParam {
     fn from(value: DisperserBlobQuorumParam) -> Self {
         Self {
@@ -88,34 +64,16 @@ pub struct BlobHeader {
     pub blob_quorum_params: Vec<BlobQuorumParam>,
 }
 
-impl BlobHeader {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        bytes.extend(self.commitment.to_bytes());
-        bytes.extend(&self.data_length.to_be_bytes());
-        bytes.extend(&self.blob_quorum_params.len().to_be_bytes());
-
-        for quorum in &self.blob_quorum_params {
-            bytes.extend(quorum.to_bytes());
-        }
-
-        bytes
-    }
-}
-
 impl TryFrom<DisperserBlobHeader> for BlobHeader {
     type Error = ConversionError;
     fn try_from(value: DisperserBlobHeader) -> Result<Self, Self::Error> {
-        if value.commitment.is_none() {
-            return Err(ConversionError::NotPresentError);
-        }
         let blob_quorum_params: Vec<BlobQuorumParam> = value
             .blob_quorum_params
             .iter()
             .map(|param| BlobQuorumParam::from(param.clone()))
             .collect();
         Ok(Self {
-            commitment: G1Commitment::from(value.commitment.unwrap()),
+            commitment: G1Commitment::from(value.commitment.ok_or(ConversionError::NotPresent)?),
             data_length: value.data_length,
             blob_quorum_params,
         })
@@ -130,21 +88,6 @@ pub struct BatchHeader {
     pub reference_block_number: u32,
 }
 
-impl BatchHeader {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        bytes.extend(&self.batch_root.len().to_be_bytes());
-        bytes.extend(&self.batch_root);
-        bytes.extend(&self.quorum_numbers.len().to_be_bytes());
-        bytes.extend(&self.quorum_numbers);
-        bytes.extend(&self.quorum_signed_percentages.len().to_be_bytes());
-        bytes.extend(&self.quorum_signed_percentages);
-        bytes.extend(&self.reference_block_number.to_be_bytes());
-
-        bytes
-    }
-}
-
 impl From<DisperserBatchHeader> for BatchHeader {
     fn from(value: DisperserBatchHeader) -> Self {
         Self {
@@ -165,25 +108,11 @@ pub struct BatchMetadata {
     pub batch_header_hash: Vec<u8>,
 }
 
-impl BatchMetadata {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        bytes.extend(self.batch_header.to_bytes());
-        bytes.extend(&self.signatory_record_hash);
-        bytes.extend(&self.confirmation_block_number.to_be_bytes());
-
-        bytes
-    }
-}
-
 impl TryFrom<DisperserBatchMetadata> for BatchMetadata {
     type Error = ConversionError;
     fn try_from(value: DisperserBatchMetadata) -> Result<Self, Self::Error> {
-        if value.batch_header.is_none() {
-            return Err(ConversionError::NotPresentError);
-        }
         Ok(Self {
-            batch_header: BatchHeader::from(value.batch_header.unwrap()),
+            batch_header: BatchHeader::from(value.batch_header.ok_or(ConversionError::NotPresent)?),
             signatory_record_hash: value.signatory_record_hash,
             fee: value.fee,
             confirmation_block_number: value.confirmation_block_number,
@@ -201,31 +130,15 @@ pub struct BlobVerificationProof {
     pub quorum_indexes: Vec<u8>,
 }
 
-impl BlobVerificationProof {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        bytes.extend(&self.batch_id.to_be_bytes());
-        bytes.extend(&self.blob_index.to_be_bytes());
-        bytes.extend(self.batch_medatada.to_bytes());
-        bytes.extend(&self.inclusion_proof.len().to_be_bytes());
-        bytes.extend(&self.inclusion_proof);
-        bytes.extend(&self.quorum_indexes.len().to_be_bytes());
-        bytes.extend(&self.quorum_indexes);
-
-        bytes
-    }
-}
-
 impl TryFrom<DisperserBlobVerificationProof> for BlobVerificationProof {
     type Error = ConversionError;
     fn try_from(value: DisperserBlobVerificationProof) -> Result<Self, Self::Error> {
-        if value.batch_metadata.is_none() {
-            return Err(ConversionError::NotPresentError);
-        }
         Ok(Self {
             batch_id: value.batch_id,
             blob_index: value.blob_index,
-            batch_medatada: BatchMetadata::try_from(value.batch_metadata.unwrap())?,
+            batch_medatada: BatchMetadata::try_from(
+                value.batch_metadata.ok_or(ConversionError::NotPresent)?,
+            )?,
             inclusion_proof: value.inclusion_proof,
             quorum_indexes: value.quorum_indexes,
         })
@@ -238,28 +151,17 @@ pub struct BlobInfo {
     pub blob_verification_proof: BlobVerificationProof,
 }
 
-impl BlobInfo {
-    pub fn to_bytes(&self) -> Vec<u8> {
-        let mut bytes = vec![];
-        let blob_header_bytes = self.blob_header.to_bytes();
-        bytes.extend(blob_header_bytes.len().to_be_bytes());
-        bytes.extend(blob_header_bytes);
-        let blob_verification_proof_bytes = self.blob_verification_proof.to_bytes();
-        bytes.extend(blob_verification_proof_bytes);
-        bytes
-    }
-}
-
 impl TryFrom<DisperserBlobInfo> for BlobInfo {
     type Error = ConversionError;
     fn try_from(value: DisperserBlobInfo) -> Result<Self, Self::Error> {
-        if value.blob_header.is_none() || value.blob_verification_proof.is_none() {
-            return Err(ConversionError::NotPresentError);
-        }
         Ok(Self {
-            blob_header: BlobHeader::try_from(value.blob_header.unwrap())?,
+            blob_header: BlobHeader::try_from(
+                value.blob_header.ok_or(ConversionError::NotPresent)?,
+            )?,
             blob_verification_proof: BlobVerificationProof::try_from(
-                value.blob_verification_proof.unwrap(),
+                value
+                    .blob_verification_proof
+                    .ok_or(ConversionError::NotPresent)?,
             )?,
         })
     }

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 {
-    async fn call(&self, input: &str) -> anyhow::Result<Option<Vec<u8>>>;
+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 {
@@ -59,9 +59,13 @@ mod tests {
 
     #[async_trait::async_trait]
     impl GetBlobData for MockGetBlobData {
-        async fn call(&self, _input: &'_ str) -> anyhow::Result<Option<Vec<u8>>> {
+        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"]
@@ -70,7 +74,7 @@ mod tests {
     async fn test_non_auth_dispersal() {
         let config = EigenConfig {
             disperser_rpc: "https://disperser-holesky.eigenda.xyz:443".to_string(),
-            settlement_layer_confirmation_depth: -1,
+            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,
@@ -110,7 +114,7 @@ mod tests {
     async fn test_auth_dispersal() {
         let config = EigenConfig {
             disperser_rpc: "https://disperser-holesky.eigenda.xyz:443".to_string(),
-            settlement_layer_confirmation_depth: -1,
+            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,