fix(eigen-client-extra-features): address comments

Cargo.toml

@@ -226,7 +226,7 @@ tokio-stream = "0.1.16"
 rust-kzg-bn254 = "0.2.1"
 ark-bn254 = "0.5.0"
 num-bigint = "0.4.6"
-serial_test = "3.1.1"
+serial_test = { version = "3.1.1", features = ["file_locks"] }
 
 # Here and below:
 # We *always* pin the latest version of protocol to disallow accidental changes in the execution logic.

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

@@ -1,5 +1,11 @@
+use std::str::FromStr;
+
 use serde::Deserialize;
-use zksync_basic_types::secrets::PrivateKey;
+use zksync_basic_types::{secrets::PrivateKey, Address, H160};
+
+/// Default address of the EigenDA service manager contract deployed on Holesky.
+const DEFAULT_EIGENDA_SVC_MANAGER_ADDRESS: &str = "0xD4A7E1Bd8015057293f0D0A557088c286942e84b";
+
 /// Configuration for the EigenDA remote disperser client.
 #[derive(Clone, Debug, PartialEq, Deserialize)]
 pub struct EigenConfig {
@@ -11,7 +17,7 @@ pub struct EigenConfig {
     /// URL of the Ethereum RPC server
     pub eigenda_eth_rpc: Option<String>,
     /// Address of the service manager contract
-    pub eigenda_svc_manager_address: String,
+    pub eigenda_svc_manager_address: Address,
     /// Wait for the blob to be finalized before returning the response
     pub wait_for_finalization: bool,
     /// Authenticated dispersal
@@ -30,7 +36,7 @@ impl Default for EigenConfig {
             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(),
+            eigenda_svc_manager_address: H160::from_str(DEFAULT_EIGENDA_SVC_MANAGER_ADDRESS).unwrap_or_default(),
             wait_for_finalization: false,
             authenticated: false,
             g1_url: "https://github.com/Layr-Labs/eigenda-proxy/raw/2fd70b99ef5bf137d7bbca3461cf9e1f2c899451/resources/g1.point".to_string(),

core/lib/env_config/src/da_client.rs

@@ -90,6 +90,9 @@ impl FromEnv for DataAvailabilitySecrets {
 
 #[cfg(test)]
 mod tests {
+    use std::str::FromStr;
+
+    use zksync_basic_types::H160;
     use zksync_config::{
         configs::{
             da_client::{
@@ -251,7 +254,7 @@ mod tests {
             DA_DISPERSER_RPC="http://localhost:8080"
             DA_SETTLEMENT_LAYER_CONFIRMATION_DEPTH=0
             DA_EIGENDA_ETH_RPC="http://localhost:8545"
-            DA_EIGENDA_SVC_MANAGER_ADDRESS="0x123"
+            DA_EIGENDA_SVC_MANAGER_ADDRESS="0x0000000000000000000000000000000000000123"
             DA_WAIT_FOR_FINALIZATION=true
             DA_AUTHENTICATED=false
             DA_G1_URL="resources1"
@@ -267,7 +270,10 @@ mod tests {
                 disperser_rpc: "http://localhost:8080".to_string(),
                 settlement_layer_confirmation_depth: 0,
                 eigenda_eth_rpc: Some("http://localhost:8545".to_string()),
-                eigenda_svc_manager_address: "0x123".to_string(),
+                eigenda_svc_manager_address: H160::from_str(
+                    "0x0000000000000000000000000000000000000123"
+                )
+                .unwrap(),
                 wait_for_finalization: true,
                 authenticated: false,
                 g1_url: "resources1".to_string(),

core/lib/protobuf_config/src/da_client.rs

@@ -10,9 +10,12 @@ use zksync_config::configs::{
 };
 use zksync_protobuf::{required, ProtoRepr};
 
-use crate::proto::{
-    da_client::{self as proto},
-    object_store as object_store_proto,
+use crate::{
+    parse_h160,
+    proto::{
+        da_client::{self as proto},
+        object_store as object_store_proto,
+    },
 };
 
 impl ProtoRepr for proto::DataAvailabilityClient {
@@ -65,8 +68,8 @@ impl ProtoRepr for proto::DataAvailabilityClient {
                 .context("settlement_layer_confirmation_depth")?,
                 eigenda_eth_rpc: required(&conf.eigenda_eth_rpc).ok().cloned(),
                 eigenda_svc_manager_address: required(&conf.eigenda_svc_manager_address)
-                    .context("eigenda_svc_manager_address")?
-                    .clone(),
+                    .and_then(|x| parse_h160(x))
+                    .context("eigenda_svc_manager_address")?,
                 wait_for_finalization: *required(&conf.wait_for_finalization)
                     .context("wait_for_finalization")?,
                 authenticated: *required(&conf.authenticated).context("authenticated")?,
@@ -116,7 +119,10 @@ impl ProtoRepr for proto::DataAvailabilityClient {
                     config.settlement_layer_confirmation_depth,
                 ),
                 eigenda_eth_rpc: config.eigenda_eth_rpc.clone(),
-                eigenda_svc_manager_address: Some(config.eigenda_svc_manager_address.clone()),
+                eigenda_svc_manager_address: Some(format!(
+                    "{:?}",
+                    config.eigenda_svc_manager_address
+                )),
                 wait_for_finalization: Some(config.wait_for_finalization),
                 authenticated: Some(config.authenticated),
                 g1_url: Some(config.g1_url.clone()),

core/node/da_clients/Cargo.toml

@@ -65,6 +65,7 @@ num-bigint.workspace = true
 zksync_web3_decl.workspace = true
 zksync_eth_client.workspace = true
 url.workspace = true
+thiserror.workspace = true
 
 [dev-dependencies]
 serial_test.workspace = true

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

@@ -15,8 +15,6 @@ use crate::utils::to_retriable_da_error;
 #[async_trait]
 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.
@@ -29,7 +27,7 @@ impl EigenClient {
     pub async fn new(
         config: EigenConfig,
         secrets: EigenSecrets,
-        get_blob_data: Box<dyn GetBlobData>,
+        get_blob_data: Arc<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))?;
@@ -80,3 +78,220 @@ impl DataAvailabilityClient for EigenClient {
         Some(RawEigenClient::blob_size_limit())
     }
 }
+
+/// EigenDA Client tests are ignored by default, because they require a remote dependency,
+/// which may not always be available, causing tests to be flaky.
+/// To run these tests, use the following command:
+/// `cargo test -p zksync_da_clients -- --ignored`
+#[cfg(test)]
+mod tests {
+    use std::{str::FromStr, sync::Arc, time::Duration};
+
+    use backon::{ConstantBuilder, Retryable};
+    use serial_test::file_serial;
+    use zksync_config::{configs::da_client::eigen::EigenSecrets, EigenConfig};
+    use zksync_da_client::{
+        types::{DAError, DispatchResponse},
+        DataAvailabilityClient,
+    };
+    use zksync_types::secrets::PrivateKey;
+
+    use crate::eigen::{blob_info::BlobInfo, EigenClient, GetBlobData};
+
+    impl EigenClient {
+        async fn get_blob_data(
+            &self,
+            blob_id: BlobInfo,
+        ) -> anyhow::Result<Option<Vec<u8>>, DAError> {
+            self.client.get_blob_data(blob_id).await
+        }
+
+        async fn get_commitment(&self, blob_id: &str) -> anyhow::Result<Option<BlobInfo>> {
+            self.client.get_commitment(blob_id).await
+        }
+    }
+
+    const STATUS_QUERY_TIMEOUT: u64 = 1800000; // 30 minutes
+    const STATUS_QUERY_INTERVAL: u64 = 5; // 5 ms
+
+    async fn get_blob_info(
+        client: &EigenClient,
+        result: &DispatchResponse,
+    ) -> anyhow::Result<BlobInfo> {
+        let blob_info = (|| async {
+            let blob_info = client.get_commitment(&result.blob_id).await?;
+            if blob_info.is_none() {
+                return Err(anyhow::anyhow!("Blob not found"));
+            }
+            Ok(blob_info.unwrap())
+        })
+        .retry(
+            &ConstantBuilder::default()
+                .with_delay(Duration::from_millis(STATUS_QUERY_INTERVAL))
+                .with_max_times((STATUS_QUERY_TIMEOUT / STATUS_QUERY_INTERVAL) as usize),
+        )
+        .when(|e| e.to_string().contains("Blob not found"))
+        .await?;
+
+        Ok(blob_info)
+    }
+
+    #[derive(Debug, Clone)]
+    struct MockGetBlobData;
+
+    #[async_trait::async_trait]
+    impl GetBlobData for MockGetBlobData {
+        async fn get_blob_data(&self, _input: &'_ str) -> anyhow::Result<Option<Vec<u8>>> {
+            Ok(None)
+        }
+    }
+
+    fn test_secrets() -> EigenSecrets {
+        EigenSecrets {
+            private_key: PrivateKey::from_str(
+                "d08aa7ae1bb5ddd46c3c2d8cdb5894ab9f54dec467233686ca42629e826ac4c6",
+            )
+            .unwrap(),
+        }
+    }
+
+    #[ignore = "depends on external RPC"]
+    #[tokio::test]
+    #[file_serial]
+    async fn test_non_auth_dispersal() {
+        let config = EigenConfig::default();
+        let secrets = test_secrets();
+        let client = EigenClient::new(config.clone(), secrets, Arc::new(MockGetBlobData))
+            .await
+            .unwrap();
+        let data = vec![1; 20];
+        let result = client.dispatch_blob(0, data.clone()).await.unwrap();
+
+        let blob_info = get_blob_info(&client, &result).await.unwrap();
+        let expected_inclusion_data = blob_info.clone().blob_verification_proof.inclusion_proof;
+        let actual_inclusion_data = client
+            .get_inclusion_data(&result.blob_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .data;
+        assert_eq!(expected_inclusion_data, actual_inclusion_data);
+        let retrieved_data = client.get_blob_data(blob_info).await.unwrap();
+        assert_eq!(retrieved_data.unwrap(), data);
+    }
+
+    #[ignore = "depends on external RPC"]
+    #[tokio::test]
+    #[file_serial]
+    async fn test_auth_dispersal() {
+        let config = EigenConfig {
+            authenticated: true,
+            ..EigenConfig::default()
+        };
+        let secrets = test_secrets();
+        let client = EigenClient::new(config.clone(), secrets, Arc::new(MockGetBlobData))
+            .await
+            .unwrap();
+        let data = vec![1; 20];
+        let result = client.dispatch_blob(0, data.clone()).await.unwrap();
+        let blob_info = get_blob_info(&client, &result).await.unwrap();
+
+        let expected_inclusion_data = blob_info.clone().blob_verification_proof.inclusion_proof;
+        let actual_inclusion_data = client
+            .get_inclusion_data(&result.blob_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .data;
+        assert_eq!(expected_inclusion_data, actual_inclusion_data);
+        let retrieved_data = client.get_blob_data(blob_info).await.unwrap();
+        assert_eq!(retrieved_data.unwrap(), data);
+    }
+
+    #[ignore = "depends on external RPC"]
+    #[tokio::test]
+    #[file_serial]
+    async fn test_wait_for_finalization() {
+        let config = EigenConfig {
+            wait_for_finalization: true,
+            authenticated: true,
+            ..EigenConfig::default()
+        };
+        let secrets = test_secrets();
+
+        let client = EigenClient::new(config.clone(), secrets, Arc::new(MockGetBlobData))
+            .await
+            .unwrap();
+        let data = vec![1; 20];
+        let result = client.dispatch_blob(0, data.clone()).await.unwrap();
+        let blob_info = get_blob_info(&client, &result).await.unwrap();
+
+        let expected_inclusion_data = blob_info.clone().blob_verification_proof.inclusion_proof;
+        let actual_inclusion_data = client
+            .get_inclusion_data(&result.blob_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .data;
+        assert_eq!(expected_inclusion_data, actual_inclusion_data);
+        let retrieved_data = client.get_blob_data(blob_info).await.unwrap();
+        assert_eq!(retrieved_data.unwrap(), data);
+    }
+
+    #[ignore = "depends on external RPC"]
+    #[tokio::test]
+    #[file_serial]
+    async fn test_settlement_layer_confirmation_depth() {
+        let config = EigenConfig {
+            settlement_layer_confirmation_depth: 5,
+            ..EigenConfig::default()
+        };
+        let secrets = test_secrets();
+        let client = EigenClient::new(config.clone(), secrets, Arc::new(MockGetBlobData))
+            .await
+            .unwrap();
+        let data = vec![1; 20];
+        let result = client.dispatch_blob(0, data.clone()).await.unwrap();
+        let blob_info = get_blob_info(&client, &result).await.unwrap();
+
+        let expected_inclusion_data = blob_info.clone().blob_verification_proof.inclusion_proof;
+        let actual_inclusion_data = client
+            .get_inclusion_data(&result.blob_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .data;
+        assert_eq!(expected_inclusion_data, actual_inclusion_data);
+        let retrieved_data = client.get_blob_data(blob_info).await.unwrap();
+        assert_eq!(retrieved_data.unwrap(), data);
+    }
+
+    #[ignore = "depends on external RPC"]
+    #[tokio::test]
+    #[file_serial]
+    async fn test_auth_dispersal_settlement_layer_confirmation_depth() {
+        let config = EigenConfig {
+            settlement_layer_confirmation_depth: 5,
+            authenticated: true,
+            ..EigenConfig::default()
+        };
+        let secrets = test_secrets();
+        let client = EigenClient::new(config.clone(), secrets, Arc::new(MockGetBlobData))
+            .await
+            .unwrap();
+        let data = vec![1; 20];
+        let result = client.dispatch_blob(0, data.clone()).await.unwrap();
+        let blob_info = get_blob_info(&client, &result).await.unwrap();
+
+        let expected_inclusion_data = blob_info.clone().blob_verification_proof.inclusion_proof;
+        let actual_inclusion_data = client
+            .get_inclusion_data(&result.blob_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .data;
+        assert_eq!(expected_inclusion_data, actual_inclusion_data);
+        let retrieved_data = client.get_blob_data(blob_info).await.unwrap();
+        assert_eq!(retrieved_data.unwrap(), data);
+    }
+}