chore: da_dispatcher refactoring

matter-labs · Dec 20, 2024 · a5cf30f · a5cf30f
1 parent ea18999
commit a5cf30f
Show file tree

Hide file tree

Showing 7 changed files with 66 additions and 30 deletions.
diff --git a/core/bin/zksync_server/src/node_builder.rs b/core/bin/zksync_server/src/node_builder.rs
@@ -530,10 +530,16 @@ impl MainNodeBuilder {
     }
 
     fn add_da_client_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 client will not be started.");
+                return Ok(self);
+            }
+        }
+
         let Some(da_client_config) = self.configs.da_client_config.clone() else {
-            tracing::warn!("No config for DA client, using the NoDA client");
-            self.node.add_layer(NoDAClientWiringLayer);
-            return Ok(self);
+            bail!("No config for DA client");
         };
 
         let secrets = try_load_config!(self.secrets.data_availability);
@@ -555,6 +561,9 @@ impl MainNodeBuilder {
                 self.node
                     .add_layer(ObjectStorageClientWiringLayer::new(config));
             }
+            (DAClientConfig::NoDA, _) => {
+                self.node.add_layer(NoDAClientWiringLayer);
+            }
             _ => bail!("invalid pair of da_client and da_secrets"),
         }
 

diff --git a/core/lib/config/src/configs/da_client/mod.rs b/core/lib/config/src/configs/da_client/mod.rs
@@ -8,13 +8,15 @@ pub const AVAIL_CLIENT_CONFIG_NAME: &str = "Avail";
 pub const CELESTIA_CLIENT_CONFIG_NAME: &str = "Celestia";
 pub const EIGEN_CLIENT_CONFIG_NAME: &str = "Eigen";
 pub const OBJECT_STORE_CLIENT_CONFIG_NAME: &str = "ObjectStore";
+pub const NO_DA_CLIENT_CONFIG_NAME: &str = "NoDA";
 
 #[derive(Debug, Clone, PartialEq)]
 pub enum DAClientConfig {
     Avail(AvailConfig),
     Celestia(CelestiaConfig),
     Eigen(EigenConfig),
     ObjectStore(ObjectStoreConfig),
+    NoDA,
 }
 
 impl From<AvailConfig> for DAClientConfig {

diff --git a/core/lib/config/src/configs/da_dispatcher.rs b/core/lib/config/src/configs/da_dispatcher.rs
@@ -3,7 +3,7 @@ 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_ROWS_TO_DISPATCH: u32 = 3;
 pub const DEFAULT_MAX_RETRIES: u16 = 5;
 pub const DEFAULT_USE_DUMMY_INCLUSION_DATA: bool = false;
 

diff --git a/core/lib/protobuf_config/src/da_client.rs b/core/lib/protobuf_config/src/da_client.rs
@@ -5,7 +5,7 @@ use zksync_config::configs::{
         avail::{AvailClientConfig, AvailConfig, AvailDefaultConfig, AvailGasRelayConfig},
         celestia::CelestiaConfig,
         eigen::EigenConfig,
-        DAClientConfig::{Avail, Celestia, Eigen, ObjectStore},
+        DAClientConfig::{Avail, Celestia, Eigen, NoDA, ObjectStore},
     },
 };
 use zksync_protobuf::{required, ProtoRepr};
@@ -62,6 +62,7 @@ impl ProtoRepr for proto::DataAvailabilityClient {
             proto::data_availability_client::Config::ObjectStore(conf) => {
                 ObjectStore(object_store_proto::ObjectStore::read(conf)?)
             }
+            proto::data_availability_client::Config::NoDa(_) => NoDA,
         };
 
         Ok(client)
@@ -102,6 +103,7 @@ impl ProtoRepr for proto::DataAvailabilityClient {
             ObjectStore(config) => proto::data_availability_client::Config::ObjectStore(
                 object_store_proto::ObjectStore::build(config),
             ),
+            NoDA => proto::data_availability_client::Config::NoDa(proto::NoDaConfig {}),
         };
 
         Self {

diff --git a/core/lib/protobuf_config/src/proto/config/da_client.proto b/core/lib/protobuf_config/src/proto/config/da_client.proto
@@ -41,12 +41,15 @@ message EigenConfig {
   optional uint64 inclusion_polling_interval_ms = 2;
 }
 
+message NoDAConfig {}
+
 message DataAvailabilityClient {
   // oneof in protobuf allows for None
   oneof config {
     AvailConfig avail = 1;
     object_store.ObjectStore object_store = 2;
     CelestiaConfig celestia = 3;
     EigenConfig eigen = 4;
+    NoDAConfig no_da = 5;
   }
 }
diff --git a/core/node/da_dispatcher/src/da_dispatcher.rs b/core/node/da_dispatcher/src/da_dispatcher.rs
@@ -1,4 +1,4 @@
-use std::{future::Future, time::Duration};
+use std::{future::Future, sync::Arc, time::Duration};
 
 use anyhow::Context;
 use chrono::Utc;
@@ -14,7 +14,7 @@ use zksync_types::L1BatchNumber;
 
 use crate::metrics::METRICS;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct DataAvailabilityDispatcher {
     client: Box<dyn DataAvailabilityClient>,
     pool: ConnectionPool<Core>,
@@ -35,37 +35,55 @@ impl DataAvailabilityDispatcher {
     }
 
     pub async fn run(self, mut stop_receiver: Receiver<bool>) -> anyhow::Result<()> {
-        loop {
-            if *stop_receiver.borrow() {
-                break;
-            }
+        let self_arc = Arc::new(self.clone());
 
-            let subtasks = futures::future::join(
-                async {
-                    if let Err(err) = self.dispatch().await {
-                        tracing::error!("dispatch error {err:?}");
-                    }
-                },
-                async {
-                    if let Err(err) = self.poll_for_inclusion().await {
-                        tracing::error!("poll_for_inclusion error {err:?}");
-                    }
-                },
-            );
+        let mut stop_receiver_dispatch = stop_receiver.clone();
 
-            tokio::select! {
-                _ = subtasks => {},
-                _ = stop_receiver.changed() => {
+        let dispatch_task = tokio::spawn(async move {
+            loop {
+                if *stop_receiver_dispatch.borrow() {
                     break;
                 }
-            }
 
-            if tokio::time::timeout(self.config.polling_interval(), stop_receiver.changed())
+                if let Err(err) = self_arc.dispatch().await {
+                    tracing::error!("dispatch error {err:?}");
+                }
+
+                if tokio::time::timeout(
+                    self_arc.config.polling_interval(),
+                    stop_receiver_dispatch.changed(),
+                )
                 .await
                 .is_ok()
-            {
-                break;
+                {
+                    break;
+                }
+            }
+        });
+
+        let inclusion_task = tokio::spawn(async move {
+            loop {
+                if *stop_receiver.borrow() {
+                    break;
+                }
+
+                if let Err(err) = self.poll_for_inclusion().await {
+                    tracing::error!("poll_for_inclusion error {err:?}");
+                }
+
+                if tokio::time::timeout(self.config.polling_interval(), stop_receiver.changed())
+                    .await
+                    .is_ok()
+                {
+                    break;
+                }
             }
+        });
+
+        tokio::select! {
+            _ = dispatch_task => {},
+            _ = inclusion_task => {},
+            _ = stop_receiver.changed() => {},
         }
 
         tracing::info!("Stop signal received, da_dispatcher is shutting down");

diff --git a/etc/env/file_based/overrides/validium.yaml b/etc/env/file_based/overrides/validium.yaml
@@ -4,3 +4,5 @@ eth:
 state_keeper:
   pubdata_overhead_part: 0
   compute_overhead_part: 1
+da_client:
+  no_da: