feat(eigenda): implement eigenDA client remaining features

Cargo.toml

@@ -222,6 +222,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 = "3.1.1"
 
 # 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

@@ -525,6 +525,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));
@@ -535,7 +536,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.expose_str().to_string());
+                }
                 self.node.add_layer(EigenWiringLayer::new(config, secret));
             }
 

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

@@ -1,10 +1,27 @@
 use serde::Deserialize;
 use zksync_basic_types::secrets::PrivateKey;
-
-#[derive(Clone, Debug, Default, PartialEq, Deserialize)]
+/// Configuration for the EigenDA remote disperser client.
+#[derive(Clone, Debug, PartialEq, Deserialize, Default)]
 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: i32,
+    /// URL of the Ethereum RPC server
+    pub eigenda_eth_rpc: Option<String>,
+    /// Address of the service manager contract
+    pub eigenda_svc_manager_address: String,
+    /// Wait for the blob to be finalized before returning the response
+    pub wait_for_finalization: bool,
+    /// Authenticated dispersal
+    pub authenticated: bool,
+    /// Url to the file containing the G1 point used for KZG
+    pub g1_url: String,
+    /// Url to the file containing the G2 point used for KZG
+    pub g2_url: String,
+    /// Chain ID of the Ethereum network
+    pub chain_id: u64,
 }
 
 #[derive(Clone, Debug, PartialEq)]

core/lib/dal/src/data_availability_dal.rs

@@ -216,4 +216,43 @@ 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
+            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
+                AND pubdata_input IS NOT NULL
+            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),
+        });
+
+        Ok(row)
+    }
 }

core/lib/env_config/src/da_client.rs

@@ -248,17 +248,31 @@ 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="0x123"
+            DA_WAIT_FOR_FINALIZATION=true
+            DA_AUTHENTICATED=false
+            DA_G1_URL="resources1"
+            DA_G2_URL="resources2"
+            DA_CHAIN_ID=1
         "#;
         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("http://localhost:8545".to_string()),
+                eigenda_svc_manager_address: "0x123".to_string(),
+                wait_for_finalization: true,
+                authenticated: false,
+                g1_url: "resources1".to_string(),
+                g2_url: "resources2".to_string(),
+                chain_id: 1
             })
         );
     }

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

@@ -10,7 +10,10 @@ use zksync_config::configs::{
 };
 use zksync_protobuf::{required, ProtoRepr};
 
-use crate::proto::{da_client as proto, object_store as object_store_proto};
+use crate::proto::{
+    da_client::{self as proto},
+    object_store as object_store_proto,
+};
 
 impl ProtoRepr for proto::DataAvailabilityClient {
     type Type = configs::DAClientConfig;
@@ -53,11 +56,26 @@ 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(),
+                settlement_layer_confirmation_depth: *required(
+                    &conf.settlement_layer_confirmation_depth,
+                )
+                .context("settlement_layer_confirmation_depth")?,
+                eigenda_eth_rpc: match required(&conf.eigenda_eth_rpc) {
+                    Ok(rpc) => Some(rpc.clone()),
+                    Err(_) => None,
+                },
+                eigenda_svc_manager_address: required(&conf.eigenda_svc_manager_address)
+                    .context("eigenda_svc_manager_address")?
                     .clone(),
-                inclusion_polling_interval_ms: *required(&conf.inclusion_polling_interval_ms)
-                    .context("inclusion_polling_interval_ms")?,
+                wait_for_finalization: *required(&conf.wait_for_finalization)
+                    .context("wait_for_finalization")?,
+                authenticated: *required(&conf.authenticated).context("authenticated")?,
+                g1_url: required(&conf.g1_url).context("g1_url")?.clone(),
+                g2_url: required(&conf.g2_url).context("g2_url")?.clone(),
+                chain_id: *required(&conf.chain_id).context("chain_id")?,
             }),
             proto::data_availability_client::Config::ObjectStore(conf) => {
                 ObjectStore(object_store_proto::ObjectStore::read(conf)?)
@@ -96,8 +114,17 @@ 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.clone(),
+                eigenda_svc_manager_address: Some(config.eigenda_svc_manager_address.clone()),
+                wait_for_finalization: Some(config.wait_for_finalization),
+                authenticated: Some(config.authenticated),
+                g1_url: Some(config.g1_url.clone()),
+                g2_url: Some(config.g2_url.clone()),
+                chain_id: Some(config.chain_id),
             }),
             ObjectStore(config) => proto::data_availability_client::Config::ObjectStore(
                 object_store_proto::ObjectStore::build(config),

core/lib/protobuf_config/src/proto/config/da_client.proto

@@ -37,8 +37,17 @@ message CelestiaConfig {
 }
 
 message EigenConfig {
-  optional string rpc_node_url = 1;
-  optional uint64 inclusion_polling_interval_ms = 2;
+  optional string disperser_rpc = 3;
+  optional int32 settlement_layer_confirmation_depth = 4;
+  optional string eigenda_eth_rpc = 5;
+  optional string eigenda_svc_manager_address = 6;
+  optional bool wait_for_finalization = 7;
+  optional bool authenticated = 8;
+  optional string g1_url = 9;
+  optional string g2_url = 10;
+  optional uint64 chain_id = 11;
+  reserved 1,2;
+  reserved "rpc_node_url","inclusion_polling_interval_ms";
 }
 
 message DataAvailabilityClient {

core/node/da_clients/Cargo.toml

@@ -55,3 +55,14 @@ pbjson-types.workspace = true
 
 # Eigen dependencies
 tokio-stream.workspace = true
+rand.workspace = true
+sha3.workspace = true
+tiny-keccak.workspace = true
+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

core/node/da_clients/src/eigen/README.md

@@ -1,11 +1,13 @@
-# EigenDA client
+# EigenDA Client
 
----
+EigenDA is as a high-throughput data availability layer for rollups. It is an EigenLayer AVS (Actively Validated
+Service), so it leverages Ethereum's economic security instead of bootstrapping a new network with its own validators.
+For more information you can check the [docs](https://docs.eigenda.xyz/).
 
-This is an implementation of the EigenDA client capable of sending the blobs to DA layer. It uses authenticated
-requests, though the auth headers are kind of mocked in the current API implementation.
+## Temporary
 
-The generated files are received by compiling the `.proto` files from EigenDA repo using the following function:
+In order to implement the integration we generated some `.proto` files from EigenDA repo that were compiled using the
+following function:
 
 ```rust
 pub fn compile_protos() {
@@ -28,8 +30,8 @@ pub fn compile_protos() {
 }
 ```
 
+The generated folder is considered a temporary solution until the EigenDA has a library with either a protogen, or
+preferably a full Rust client implementation.
+
 proto files are not included here to not create confusion in case they are not updated in time, so the EigenDA
 [repo](https://github.com/Layr-Labs/eigenda/tree/master/api/proto) has to be a source of truth for the proto files.
-
-The generated folder here is considered a temporary solution until the EigenDA has a library with either a protogen, or
-preferably a full Rust client implementation.