feat: Validium with DA (matter-labs#2010)

## What ❔

This PR adds an opportunity to use external DA layers to submit the
pubdata.
The implementations of the clients will be created in this repo:
https://github.com/matter-labs/hyperchain-da
This is only the first stage, and doesn't include any modifications to
the `commitBatches` or bootloader memory, those will be added as a
follow-up.
Design doc for the feature (might be a bit outdated, but main concepts
are valid):
https://matterlabs.notion.site/Validium-with-DA-EXTERNAL-8deccba433be4ff88592a3b0f8774062
It is assumed that the pubdata publishing is subsidized by the operators
at this points, it shouldn't be a concern because:
- the DA layers' storage costs are quite low
- for most DA layers, the blob size is around 2MB (or they work on
increasing it to be like that), so we could in theory fit ~30-35k txs
there, which makes that cost even lower per tx

## Why ❔

We have many partners who want to use the different DA layers, this PR
provides a common interface for all of them

## Checklist

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zk fmt` and `zk lint`.
- [ ] Spellcheck has been run via `zk spellcheck`.

.github/workflows/ci-core-reusable.yml

@@ -135,7 +135,7 @@ jobs:
         base_token: ["Eth", "Custom"]
         deployment_mode: ["Rollup", "Validium"]
     env:
-      SERVER_COMPONENTS: "api,tree,eth,state_keeper,housekeeper,commitment_generator,vm_runner_protective_reads${{ matrix.consensus && ',consensus' || '' }}"
+      SERVER_COMPONENTS: "api,tree,eth,state_keeper,housekeeper,commitment_generator,vm_runner_protective_reads,da_dispatcher${{ matrix.consensus && ',consensus' || '' }}"
 
     runs-on: [matterlabs-ci-runner]
     steps:
@@ -288,6 +288,10 @@ jobs:
         if: always()
         run: ci_run cat core/tests/upgrade-test/upgrade.log || true
 
+      - name: Show fee-projection.log logs
+        if: always()
+        run: ci_run cat core/tests/ts-integration/fees.log || true
+
       - name: Show sccache logs
         if: always()
         run: |
@@ -305,7 +309,7 @@ jobs:
     runs-on: [matterlabs-ci-runner]
 
     env:
-      SERVER_COMPONENTS: "api,tree,eth,state_keeper,housekeeper,commitment_generator,vm_runner_protective_reads${{ matrix.consensus && ',consensus' || '' }}"
+      SERVER_COMPONENTS: "api,tree,eth,state_keeper,housekeeper,commitment_generator,vm_runner_protective_reads,da_dispatcher${{ matrix.consensus && ',consensus' || '' }}"
       EXT_NODE_FLAGS: "${{ matrix.consensus && '-- --enable-consensus' || '' }}"
 
     steps:

Cargo.toml

@@ -20,6 +20,7 @@ members = [
     "core/node/shared_metrics",
     "core/node/db_pruner",
     "core/node/fee_model",
+    "core/node/da_dispatcher",
     "core/node/eth_sender",
     "core/node/vm_runner",
     "core/node/test_utils",
@@ -44,6 +45,8 @@ members = [
     "core/lib/circuit_breaker",
     "core/lib/dal",
     "core/lib/env_config",
+    "core/lib/da_client",
+    "core/lib/default_da_clients",
     "core/lib/eth_client",
     "core/lib/eth_signer",
     "core/lib/l1_contract_interface",
@@ -223,6 +226,8 @@ zksync_dal = { path = "core/lib/dal" }
 zksync_db_connection = { path = "core/lib/db_connection" }
 zksync_env_config = { path = "core/lib/env_config" }
 zksync_eth_client = { path = "core/lib/eth_client" }
+zksync_da_client = { path = "core/lib/da_client" }
+zksync_default_da_clients = { path = "core/lib/default_da_clients" }
 zksync_eth_signer = { path = "core/lib/eth_signer" }
 zksync_health_check = { path = "core/lib/health_check" }
 zksync_l1_contract_interface = { path = "core/lib/l1_contract_interface" }
@@ -254,6 +259,7 @@ zksync_block_reverter = { path = "core/node/block_reverter" }
 zksync_commitment_generator = { path = "core/node/commitment_generator" }
 zksync_house_keeper = { path = "core/node/house_keeper" }
 zksync_node_genesis = { path = "core/node/genesis" }
+zksync_da_dispatcher = { path = "core/node/da_dispatcher" }
 zksync_eth_sender = { path = "core/node/eth_sender" }
 zksync_node_db_pruner = { path = "core/node/db_pruner" }
 zksync_node_fee_model = { path = "core/node/fee_model" }

core/bin/zksync_server/Cargo.toml

@@ -20,6 +20,7 @@ zksync_utils.workspace = true
 zksync_types.workspace = true
 zksync_core_leftovers.workspace = true
 zksync_node_genesis.workspace = true
+zksync_default_da_clients.workspace = true
 
 # Consensus dependenices
 zksync_consensus_crypto.workspace = true

core/bin/zksync_server/src/main.rs

@@ -16,8 +16,8 @@ use zksync_config::{
         L1Secrets, ObservabilityConfig, PrometheusConfig, ProofDataHandlerConfig,
         ProtectiveReadsWriterConfig, Secrets,
     },
-    ApiConfig, ContractVerifierConfig, DBConfig, EthConfig, EthWatchConfig, GasAdjusterConfig,
-    GenesisConfig, ObjectStoreConfig, PostgresConfig, SnapshotsCreatorConfig,
+    ApiConfig, ContractVerifierConfig, DADispatcherConfig, DBConfig, EthConfig, EthWatchConfig,
+    GasAdjusterConfig, GenesisConfig, ObjectStoreConfig, PostgresConfig, SnapshotsCreatorConfig,
 };
 use zksync_core_leftovers::{
     genesis_init, is_genesis_needed,
@@ -47,7 +47,7 @@ struct Cli {
     /// Comma-separated list of components to launch.
     #[arg(
         long,
-        default_value = "api,tree,eth,state_keeper,housekeeper,tee_verifier_input_producer,commitment_generator"
+        default_value = "api,tree,eth,state_keeper,housekeeper,tee_verifier_input_producer,commitment_generator,da_dispatcher"
     )]
     components: ComponentsToRun,
     /// Path to the yaml config. If set, it will be used instead of env vars.
@@ -268,6 +268,7 @@ fn load_env_config() -> anyhow::Result<TempConfigStore> {
         gas_adjuster_config: GasAdjusterConfig::from_env().ok(),
         observability: ObservabilityConfig::from_env().ok(),
         snapshot_creator: SnapshotsCreatorConfig::from_env().ok(),
+        da_dispatcher_config: DADispatcherConfig::from_env().ok(),
         protective_reads_writer_config: ProtectiveReadsWriterConfig::from_env().ok(),
         core_object_store: ObjectStoreConfig::from_env().ok(),
         commitment_generator: None,

core/bin/zksync_server/src/node_builder.rs

@@ -3,10 +3,17 @@
 
 use anyhow::Context;
 use zksync_config::{
-    configs::{consensus::ConsensusConfig, wallets::Wallets, GeneralConfig, Secrets},
+    configs::{
+        consensus::ConsensusConfig, eth_sender::PubdataSendingMode, wallets::Wallets,
+        GeneralConfig, Secrets,
+    },
     ContractsConfig, GenesisConfig,
 };
 use zksync_core_leftovers::Component;
+use zksync_default_da_clients::{
+    no_da::wiring_layer::NoDAClientWiringLayer,
+    object_store::{config::DAObjectStoreConfig, wiring_layer::ObjectStorageClientWiringLayer},
+};
 use zksync_metadata_calculator::MetadataCalculatorConfig;
 use zksync_node_api_server::{
     tx_sender::{ApiContracts, TxSenderConfig},
@@ -18,6 +25,7 @@ use zksync_node_framework::{
         commitment_generator::CommitmentGeneratorLayer,
         consensus::{ConsensusLayer, Mode as ConsensusMode},
         contract_verification_api::ContractVerificationApiLayer,
+        da_dispatcher::DataAvailabilityDispatcherLayer,
         eth_sender::{EthTxAggregatorLayer, EthTxManagerLayer},
         eth_watch::EthWatchLayer,
         healtcheck_server::HealthCheckLayer,
@@ -444,6 +452,38 @@ impl MainNodeBuilder {
         Ok(self)
     }
 
+    fn add_no_da_client_layer(mut self) -> anyhow::Result<Self> {
+        self.node.add_layer(NoDAClientWiringLayer);
+        Ok(self)
+    }
+
+    #[allow(dead_code)]
+    fn add_object_storage_da_client_layer(mut self) -> anyhow::Result<Self> {
+        let object_store_config = DAObjectStoreConfig::from_env()?;
+        self.node
+            .add_layer(ObjectStorageClientWiringLayer::new(object_store_config.0));
+        Ok(self)
+    }
+
+    fn add_da_dispatcher_layer(mut self) -> anyhow::Result<Self> {
+        let eth_sender_config = try_load_config!(self.configs.eth);
+        if let Some(sender_config) = eth_sender_config.sender {
+            if sender_config.pubdata_sending_mode != PubdataSendingMode::Custom {
+                tracing::warn!("DA dispatcher is enabled, but the pubdata sending mode is not `Custom`. DA dispatcher will not be started.");
+                return Ok(self);
+            }
+        }
+
+        let state_keeper_config = try_load_config!(self.configs.state_keeper_config);
+        let da_config = try_load_config!(self.configs.da_dispatcher_config);
+        self.node.add_layer(DataAvailabilityDispatcherLayer::new(
+            state_keeper_config,
+            da_config,
+        ));
+
+        Ok(self)
+    }
+
     fn add_vm_runner_protective_reads_layer(mut self) -> anyhow::Result<Self> {
         let protective_reads_writer_config =
             try_load_config!(self.configs.protective_reads_writer_config);
@@ -539,6 +579,9 @@ impl MainNodeBuilder {
                 Component::CommitmentGenerator => {
                     self = self.add_commitment_generator_layer()?;
                 }
+                Component::DADispatcher => {
+                    self = self.add_no_da_client_layer()?.add_da_dispatcher_layer()?;
+                }
                 Component::VmRunnerProtectiveReads => {
                     self = self.add_vm_runner_protective_reads_layer()?;
                 }

core/lib/config/src/configs/chain.rs

@@ -105,7 +105,12 @@ pub struct StateKeeperConfig {
     pub batch_overhead_l1_gas: u64,
     /// The maximum amount of gas that can be used by the batch. This value is derived from the circuits limitation per batch.
     pub max_gas_per_batch: u64,
-    /// The maximum amount of pubdata that can be used by the batch. Note that if the calldata is used as pubdata, this variable should not exceed 128kb.
+    /// The maximum amount of pubdata that can be used by the batch.
+    /// This variable should not exceed:
+    /// - 128kb for calldata-based rollups
+    /// - 120kb * n, where `n` is a number of blobs for blob-based rollups
+    /// - the DA layer's blob size limit for the DA layer-based validiums
+    /// - 100 MB for the object store-based or no-da validiums
     pub max_pubdata_per_batch: u64,
 
     /// The version of the fee model to use.

core/lib/config/src/configs/da_dispatcher.rs

@@ -0,0 +1,43 @@
+use std::time::Duration;
+
+use serde::Deserialize;
+
+pub const DEFAULT_POLLING_INTERVAL_MS: u32 = 5000;
+pub const DEFAULT_MAX_ROWS_TO_DISPATCH: u32 = 100;
+pub const DEFAULT_MAX_RETRIES: u16 = 5;
+
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+pub struct DADispatcherConfig {
+    /// The interval between the `da_dispatcher's` iterations.
+    pub polling_interval_ms: Option<u32>,
+    /// The maximum number of rows to query from the database in a single query.
+    pub max_rows_to_dispatch: Option<u32>,
+    /// The maximum number of retries for the dispatch of a blob.
+    pub max_retries: Option<u16>,
+}
+
+impl DADispatcherConfig {
+    pub fn for_tests() -> Self {
+        Self {
+            polling_interval_ms: Some(DEFAULT_POLLING_INTERVAL_MS),
+            max_rows_to_dispatch: Some(DEFAULT_MAX_ROWS_TO_DISPATCH),
+            max_retries: Some(DEFAULT_MAX_RETRIES),
+        }
+    }
+
+    pub fn polling_interval(&self) -> Duration {
+        match self.polling_interval_ms {
+            Some(interval) => Duration::from_millis(interval as u64),
+            None => Duration::from_millis(DEFAULT_POLLING_INTERVAL_MS as u64),
+        }
+    }
+
+    pub fn max_rows_to_dispatch(&self) -> u32 {
+        self.max_rows_to_dispatch
+            .unwrap_or(DEFAULT_MAX_ROWS_TO_DISPATCH)
+    }
+
+    pub fn max_retries(&self) -> u16 {
+        self.max_retries.unwrap_or(DEFAULT_MAX_RETRIES)
+    }
+}

core/lib/config/src/configs/eth_sender.rs

@@ -81,6 +81,7 @@ pub enum PubdataSendingMode {
     #[default]
     Calldata,
     Blobs,
+    Custom,
 }
 
 #[derive(Debug, Deserialize, Clone, PartialEq)]
@@ -114,7 +115,7 @@ pub struct SenderConfig {
     // Max acceptable fee for sending tx it acts as a safeguard to prevent sending tx with very high fees.
     pub max_acceptable_priority_fee_in_gwei: u64,
 
-    /// The mode in which we send pubdata, either Calldata or Blobs
+    /// The mode in which we send pubdata: Calldata, Blobs or Custom (DA layers, Object Store, etc.)
     pub pubdata_sending_mode: PubdataSendingMode,
 }
 

core/lib/config/src/configs/general.rs

@@ -1,6 +1,7 @@
 use crate::{
     configs::{
         chain::{CircuitBreakerConfig, MempoolConfig, OperationsManagerConfig, StateKeeperConfig},
+        da_dispatcher::DADispatcherConfig,
         fri_prover_group::FriProverGroupConfig,
         house_keeper::HouseKeeperConfig,
         pruning::PruningConfig,
@@ -36,6 +37,7 @@ pub struct GeneralConfig {
     pub eth: Option<EthConfig>,
     pub snapshot_creator: Option<SnapshotsCreatorConfig>,
     pub observability: Option<ObservabilityConfig>,
+    pub da_dispatcher_config: Option<DADispatcherConfig>,
     pub protective_reads_writer_config: Option<ProtectiveReadsWriterConfig>,
     pub commitment_generator: Option<CommitmentGeneratorConfig>,
     pub snapshot_recovery: Option<SnapshotRecoveryConfig>,

core/lib/config/src/configs/mod.rs

@@ -4,6 +4,7 @@ pub use self::{
     commitment_generator::CommitmentGeneratorConfig,
     contract_verifier::ContractVerifierConfig,
     contracts::{ContractsConfig, EcosystemContracts},
+    da_dispatcher::DADispatcherConfig,
     database::{DBConfig, PostgresConfig},
     eth_sender::{EthConfig, GasAdjusterConfig},
     eth_watch::EthWatchConfig,
@@ -32,6 +33,7 @@ mod commitment_generator;
 pub mod consensus;
 pub mod contract_verifier;
 pub mod contracts;
+pub mod da_dispatcher;
 pub mod database;
 pub mod en_config;
 pub mod eth_sender;

core/lib/config/src/lib.rs

@@ -1,8 +1,9 @@
 #![allow(clippy::upper_case_acronyms, clippy::derive_partial_eq_without_eq)]
 
 pub use crate::configs::{
-    ApiConfig, ContractVerifierConfig, ContractsConfig, DBConfig, EthConfig, EthWatchConfig,
-    GasAdjusterConfig, GenesisConfig, ObjectStoreConfig, PostgresConfig, SnapshotsCreatorConfig,
+    ApiConfig, ContractVerifierConfig, ContractsConfig, DADispatcherConfig, DBConfig, EthConfig,
+    EthWatchConfig, GasAdjusterConfig, GenesisConfig, ObjectStoreConfig, PostgresConfig,
+    SnapshotsCreatorConfig,
 };
 
 pub mod configs;

core/lib/config/src/testonly.rs

@@ -350,9 +350,10 @@ impl Distribution<configs::eth_sender::ProofLoadingMode> for EncodeDist {
 impl Distribution<configs::eth_sender::PubdataSendingMode> for EncodeDist {
     fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::eth_sender::PubdataSendingMode {
         type T = configs::eth_sender::PubdataSendingMode;
-        match rng.gen_range(0..2) {
+        match rng.gen_range(0..3) {
             0 => T::Calldata,
-            _ => T::Blobs,
+            1 => T::Blobs,
+            _ => T::Custom,
         }
     }
 }