feat(eigenda): implement eigenDA client remaining features

core/Cargo.toml

@@ -225,6 +225,10 @@ pbjson-types = "0.6.0"
 
 # Eigen
 tokio-stream = "0.1.16"
+rust-kzg-bn254 = "0.2.1"
+ark-bn254 = "0.5.0"
+num-bigint = "0.4.6"
+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/bin/zksync_server/src/node_builder.rs

@@ -569,6 +569,7 @@ impl MainNodeBuilder {
         }
 
         let secrets = try_load_config!(self.secrets.data_availability);
+        let l1_secrets = try_load_config!(self.secrets.l1);
         match (da_client_config, secrets) {
             (DAClientConfig::Avail(config), DataAvailabilitySecrets::Avail(secret)) => {
                 self.node.add_layer(AvailWiringLayer::new(config, secret));
@@ -579,7 +580,10 @@ impl MainNodeBuilder {
                     .add_layer(CelestiaWiringLayer::new(config, secret));
             }
 
-            (DAClientConfig::Eigen(config), DataAvailabilitySecrets::Eigen(secret)) => {
+            (DAClientConfig::Eigen(mut config), DataAvailabilitySecrets::Eigen(secret)) => {
+                if config.eigenda_eth_rpc.is_none() {
+                    config.eigenda_eth_rpc = Some(l1_secrets.l1_rpc_url);
+                }
                 self.node.add_layer(EigenWiringLayer::new(config, secret));
             }
 

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

@@ -1,10 +1,31 @@
 use serde::Deserialize;
-use zksync_basic_types::secrets::PrivateKey;
+use zksync_basic_types::{secrets::PrivateKey, url::SensitiveUrl, Address};
 
-#[derive(Clone, Debug, Default, PartialEq, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Deserialize)]
+pub enum PointsSource {
+    Path(String),
+    /// g1_url, g2_url
+    Url((String, String)),
+}
+
+/// Configuration for the EigenDA remote disperser client.
+#[derive(Clone, Debug, PartialEq, Deserialize)]
 pub struct EigenConfig {
-    pub rpc_node_url: String,
-    pub inclusion_polling_interval_ms: u64,
+    /// 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: u32,
+    /// URL of the Ethereum RPC server
+    pub eigenda_eth_rpc: Option<SensitiveUrl>,
+    /// Address of the service manager contract
+    pub eigenda_svc_manager_address: Address,
+    /// Wait for the blob to be finalized before returning the response
+    pub wait_for_finalization: bool,
+    /// Authenticated dispersal
+    pub authenticated: bool,
+    /// Points source
+    pub points_source: PointsSource,
 }
 
 #[derive(Clone, Debug, PartialEq)]

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/dal/src/data_availability_dal.rs

@@ -219,4 +219,44 @@ impl DataAvailabilityDal<'_, '_> {
             })
             .collect())
     }
+
+    /// Fetches the pubdata for the L1 batch with a given blob id.
+    pub async fn get_blob_data_by_blob_id(
+        &mut self,
+        blob_id: &str,
+    ) -> DalResult<Option<L1BatchDA>> {
+        let row = sqlx::query!(
+            r#"
+            SELECT
+                number,
+                pubdata_input,
+                sealed_at
+            FROM
+                l1_batches
+            LEFT JOIN
+                data_availability
+                ON data_availability.l1_batch_number = l1_batches.number
+            WHERE
+                number != 0
+                AND data_availability.blob_id = $1
+            ORDER BY
+                number
+            LIMIT
+                1
+            "#,
+            blob_id,
+        )
+        .instrument("get_blob_data_by_blob_id")
+        .with_arg("blob_id", &blob_id)
+        .fetch_optional(self.storage)
+        .await?
+        .map(|row| L1BatchDA {
+            // `unwrap` is safe here because we have a `WHERE` clause that filters out `NULL` values
+            pubdata: row.pubdata_input.unwrap(),
+            l1_batch_number: L1BatchNumber(row.number as u32),
+            sealed_at: row.sealed_at.unwrap().and_utc(),
+        });
+
+        Ok(row)
+    }
 }

core/lib/env_config/src/da_client.rs

@@ -1,17 +1,22 @@
-use std::env;
+use std::{env, str::FromStr};
 
-use zksync_config::configs::{
-    da_client::{
-        avail::{
-            AvailClientConfig, AvailSecrets, AVAIL_FULL_CLIENT_NAME, AVAIL_GAS_RELAY_CLIENT_NAME,
+use zksync_basic_types::{url::SensitiveUrl, H160};
+use zksync_config::{
+    configs::{
+        da_client::{
+            avail::{
+                AvailClientConfig, AvailSecrets, AVAIL_FULL_CLIENT_NAME,
+                AVAIL_GAS_RELAY_CLIENT_NAME,
+            },
+            celestia::CelestiaSecrets,
+            eigen::EigenSecrets,
+            DAClientConfig, AVAIL_CLIENT_CONFIG_NAME, CELESTIA_CLIENT_CONFIG_NAME,
+            EIGEN_CLIENT_CONFIG_NAME, NO_DA_CLIENT_CONFIG_NAME, OBJECT_STORE_CLIENT_CONFIG_NAME,
         },
-        celestia::CelestiaSecrets,
-        eigen::EigenSecrets,
-        DAClientConfig, AVAIL_CLIENT_CONFIG_NAME, CELESTIA_CLIENT_CONFIG_NAME,
-        EIGEN_CLIENT_CONFIG_NAME, NO_DA_CLIENT_CONFIG_NAME, OBJECT_STORE_CLIENT_CONFIG_NAME,
+        secrets::DataAvailabilitySecrets,
+        AvailConfig,
     },
-    secrets::DataAvailabilitySecrets,
-    AvailConfig,
+    EigenConfig,
 };
 
 use crate::{envy_load, FromEnv};
@@ -34,7 +39,35 @@ impl FromEnv for DAClientConfig {
                 },
             }),
             CELESTIA_CLIENT_CONFIG_NAME => Self::Celestia(envy_load("da_celestia_config", "DA_")?),
-            EIGEN_CLIENT_CONFIG_NAME => Self::Eigen(envy_load("da_eigen_config", "DA_")?),
+            EIGEN_CLIENT_CONFIG_NAME => Self::Eigen(EigenConfig {
+                disperser_rpc: env::var("DA_DISPERSER_RPC")?,
+                settlement_layer_confirmation_depth: env::var(
+                    "DA_SETTLEMENT_LAYER_CONFIRMATION_DEPTH",
+                )?
+                .parse()?,
+                eigenda_eth_rpc: match env::var("DA_EIGENDA_ETH_RPC") {
+                    // Use a specific L1 RPC URL for the EigenDA client.
+                    Ok(url) => Some(SensitiveUrl::from_str(&url)?),
+                    // Err means that the environment variable is not set.
+                    // Use zkSync default L1 RPC for the EigenDA client.
+                    Err(_) => None,
+                },
+                eigenda_svc_manager_address: H160::from_str(&env::var(
+                    "DA_EIGENDA_SVC_MANAGER_ADDRESS",
+                )?)?,
+                wait_for_finalization: env::var("DA_WAIT_FOR_FINALIZATION")?.parse()?,
+                authenticated: env::var("DA_AUTHENTICATED")?.parse()?,
+                points_source: match env::var("DA_POINTS_SOURCE")?.as_str() {
+                    "Path" => zksync_config::configs::da_client::eigen::PointsSource::Path(
+                        env::var("DA_POINTS_PATH")?,
+                    ),
+                    "Url" => zksync_config::configs::da_client::eigen::PointsSource::Url((
+                        env::var("DA_POINTS_LINK_G1")?,
+                        env::var("DA_POINTS_LINK_G2")?,
+                    )),
+                    _ => anyhow::bail!("Unknown Eigen points type"),
+                },
+            }),
             OBJECT_STORE_CLIENT_CONFIG_NAME => {
                 Self::ObjectStore(envy_load("da_object_store", "DA_")?)
             }
@@ -87,10 +120,14 @@ impl FromEnv for DataAvailabilitySecrets {
 
 #[cfg(test)]
 mod tests {
+    use std::str::FromStr;
+
+    use zksync_basic_types::url::SensitiveUrl;
     use zksync_config::{
         configs::{
             da_client::{
                 avail::{AvailClientConfig, AvailDefaultConfig},
+                eigen::PointsSource,
                 DAClientConfig::{self, ObjectStore},
             },
             object_store::ObjectStoreMode::GCS,
@@ -244,17 +281,30 @@ mod tests {
         let mut lock = MUTEX.lock();
         let config = r#"
             DA_CLIENT="Eigen"
-            DA_RPC_NODE_URL="localhost:12345"
-            DA_INCLUSION_POLLING_INTERVAL_MS="1000"
+            DA_DISPERSER_RPC="http://localhost:8080"
+            DA_SETTLEMENT_LAYER_CONFIRMATION_DEPTH=0
+            DA_EIGENDA_ETH_RPC="http://localhost:8545"
+            DA_EIGENDA_SVC_MANAGER_ADDRESS="0x0000000000000000000000000000000000000123"
+            DA_WAIT_FOR_FINALIZATION=true
+            DA_AUTHENTICATED=false
+            DA_POINTS_SOURCE="Path"
+            DA_POINTS_PATH="resources"
         "#;
         lock.set_env(config);
 
         let actual = DAClientConfig::from_env().unwrap();
         assert_eq!(
             actual,
             DAClientConfig::Eigen(EigenConfig {
-                rpc_node_url: "localhost:12345".to_string(),
-                inclusion_polling_interval_ms: 1000,
+                disperser_rpc: "http://localhost:8080".to_string(),
+                settlement_layer_confirmation_depth: 0,
+                eigenda_eth_rpc: Some(SensitiveUrl::from_str("http://localhost:8545").unwrap()),
+                eigenda_svc_manager_address: "0x0000000000000000000000000000000000000123"
+                    .parse()
+                    .unwrap(),
+                wait_for_finalization: true,
+                authenticated: false,
+                points_source: PointsSource::Path("resources".to_string()),
             })
         );
     }

core/lib/env_config/src/lib.rs

@@ -1,4 +1,3 @@
-use anyhow::Context as _;
 use serde::de::DeserializeOwned;
 
 mod api;
@@ -44,5 +43,5 @@ pub trait FromEnv: Sized {
 pub fn envy_load<T: DeserializeOwned>(name: &str, prefix: &str) -> anyhow::Result<T> {
     envy::prefixed(prefix)
         .from_env()
-        .with_context(|| format!("Cannot load config <{name}>"))
+        .map_err(|e| anyhow::anyhow!("Failed to load {} from env: {}", name, e))
 }

core/lib/protobuf_config/src/da_client.rs

@@ -1,3 +1,5 @@
+use std::str::FromStr;
+
 use anyhow::Context;
 use zksync_config::configs::{
     self,
@@ -9,8 +11,15 @@ use zksync_config::configs::{
     },
 };
 use zksync_protobuf::{required, ProtoRepr};
+use zksync_types::url::SensitiveUrl;
 
-use crate::proto::{da_client 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 {
     type Type = configs::DAClientConfig;
@@ -53,11 +62,36 @@ impl ProtoRepr for proto::DataAvailabilityClient {
                 timeout_ms: *required(&conf.timeout_ms).context("timeout_ms")?,
             }),
             proto::data_availability_client::Config::Eigen(conf) => Eigen(EigenConfig {
-                rpc_node_url: required(&conf.rpc_node_url)
-                    .context("rpc_node_url")?
+                disperser_rpc: required(&conf.disperser_rpc)
+                    .context("disperser_rpc")?
                     .clone(),
-                inclusion_polling_interval_ms: *required(&conf.inclusion_polling_interval_ms)
-                    .context("inclusion_polling_interval_ms")?,
+                settlement_layer_confirmation_depth: *required(
+                    &conf.settlement_layer_confirmation_depth,
+                )
+                .context("settlement_layer_confirmation_depth")?,
+                eigenda_eth_rpc: Some(SensitiveUrl::from_str(
+                    required(&conf.eigenda_eth_rpc).context("eigenda_eth_rpc")?,
+                )?),
+                eigenda_svc_manager_address: required(&conf.eigenda_svc_manager_address)
+                    .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")?,
+                points_source: match conf.points_source.clone() {
+                    Some(proto::eigen_config::PointsSource::Path(path)) => {
+                        zksync_config::configs::da_client::eigen::PointsSource::Path(path)
+                    }
+                    Some(proto::eigen_config::PointsSource::Url(url)) => {
+                        let g1_url = required(&url.g1_url).context("g1_url")?;
+                        let g2_url = required(&url.g2_url).context("g2_url")?;
+                        zksync_config::configs::da_client::eigen::PointsSource::Url((
+                            g1_url.to_owned(),
+                            g2_url.to_owned(),
+                        ))
+                    }
+                    None => return Err(anyhow::anyhow!("Invalid Eigen DA configuration")),
+                },
             }),
             proto::data_availability_client::Config::ObjectStore(conf) => {
                 ObjectStore(object_store_proto::ObjectStore::read(conf)?)
@@ -98,8 +132,32 @@ impl ProtoRepr for proto::DataAvailabilityClient {
                 })
             }
             Eigen(config) => proto::data_availability_client::Config::Eigen(proto::EigenConfig {
-                rpc_node_url: Some(config.rpc_node_url.clone()),
-                inclusion_polling_interval_ms: Some(config.inclusion_polling_interval_ms),
+                disperser_rpc: Some(config.disperser_rpc.clone()),
+                settlement_layer_confirmation_depth: Some(
+                    config.settlement_layer_confirmation_depth,
+                ),
+                eigenda_eth_rpc: config
+                    .eigenda_eth_rpc
+                    .as_ref()
+                    .map(|a| a.expose_str().to_string()),
+                eigenda_svc_manager_address: Some(format!(
+                    "{:?}",
+                    config.eigenda_svc_manager_address
+                )),
+                wait_for_finalization: Some(config.wait_for_finalization),
+                authenticated: Some(config.authenticated),
+                points_source: Some(match &config.points_source {
+                    zksync_config::configs::da_client::eigen::PointsSource::Path(path) => {
+                        proto::eigen_config::PointsSource::Path(path.clone())
+                    }
+                    zksync_config::configs::da_client::eigen::PointsSource::Url((
+                        g1_url,
+                        g2_url,
+                    )) => proto::eigen_config::PointsSource::Url(proto::Url {
+                        g1_url: Some(g1_url.clone()),
+                        g2_url: Some(g2_url.clone()),
+                    }),
+                }),
             }),
             ObjectStore(config) => proto::data_availability_client::Config::ObjectStore(
                 object_store_proto::ObjectStore::build(config),