feat: new da_dispatcher metrics

core/lib/da_client/src/lib.rs

@@ -23,6 +23,8 @@ pub trait DataAvailabilityClient: Sync + Send + fmt::Debug {
 
     /// Returns the maximum size of the blob (in bytes) that can be dispatched. None means no limit.
     fn blob_size_limit(&self) -> Option<usize>;
+
+    async fn balance(&self) -> Result<u64, DAError>;
 }
 
 impl Clone for Box<dyn DataAvailabilityClient> {

core/lib/dal/src/data_availability_dal.rs

@@ -184,7 +184,8 @@ impl DataAvailabilityDal<'_, '_> {
             r#"
             SELECT
                 number,
-                pubdata_input
+                pubdata_input,
+                sealed_at
             FROM
                 l1_batches
             LEFT JOIN
@@ -195,6 +196,7 @@ impl DataAvailabilityDal<'_, '_> {
                 AND number != 0
                 AND data_availability.blob_id IS NULL
                 AND pubdata_input IS NOT NULL
+                AND sealed_at IS NOT NULL
             ORDER BY
                 number
             LIMIT
@@ -213,6 +215,7 @@ impl DataAvailabilityDal<'_, '_> {
                 // `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(),
             })
             .collect())
     }

core/lib/dal/src/models/storage_data_availability.rs

@@ -1,4 +1,4 @@
-use chrono::NaiveDateTime;
+use chrono::{DateTime, NaiveDateTime, Utc};
 use zksync_types::{pubdata_da::DataAvailabilityBlob, L1BatchNumber};
 
 /// Represents a blob in the data availability layer.
@@ -26,4 +26,5 @@ impl From<StorageDABlob> for DataAvailabilityBlob {
 pub struct L1BatchDA {
     pub pubdata: Vec<u8>,
     pub l1_batch_number: L1BatchNumber,
+    pub sealed_at: DateTime<Utc>,
 }

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

@@ -231,4 +231,27 @@ impl DataAvailabilityClient for AvailClient {
     fn blob_size_limit(&self) -> Option<usize> {
         Some(RawAvailClient::MAX_BLOB_SIZE)
     }
+
+    async fn balance(&self) -> Result<u64, DAError> {
+        match self.sdk_client.as_ref() {
+            AvailClientMode::Default(client) => {
+                let AvailClientConfig::FullClient(default_config) = &self.config.config else {
+                    unreachable!(); // validated in protobuf config
+                };
+
+                let ws_client = WsClientBuilder::default()
+                    .build(default_config.api_node_url.clone().as_str())
+                    .await
+                    .map_err(to_non_retriable_da_error)?;
+
+                Ok(client
+                    .balance(&ws_client)
+                    .await
+                    .map_err(to_non_retriable_da_error)?)
+            }
+            AvailClientMode::GasRelay(_) => {
+                Ok(0) // TODO: implement balance for gas relay
+            }
+        }
+    }
 }

core/node/da_clients/src/avail/sdk.rs

@@ -3,6 +3,7 @@
 
 use std::{fmt::Debug, sync::Arc, time};
 
+use anyhow::Context;
 use backon::{ConstantBuilder, Retryable};
 use bytes::Bytes;
 use jsonrpsee::{
@@ -22,7 +23,6 @@ use crate::utils::to_non_retriable_da_error;
 
 const PROTOCOL_VERSION: u8 = 4;
 
-/// An implementation of the `DataAvailabilityClient` trait that interacts with the Avail network.
 #[derive(Debug, Clone)]
 pub(crate) struct RawAvailClient {
     app_id: u32,
@@ -344,6 +344,23 @@ impl RawAvailClient {
 
         Ok(tx_id)
     }
+
+    /// Returns a balance of the address controlled by the `keypair`
+    pub async fn balance(&self, client: &Client) -> anyhow::Result<u64> {
+        let address = to_addr(self.keypair.clone());
+        let resp: serde_json::Value = client
+            .request("state_getStorage", rpc_params![address])
+            .await
+            .context("Error calling state_getStorage RPC")?;
+
+        let balance = resp
+            .as_str()
+            .ok_or_else(|| anyhow::anyhow!("Invalid balance"))?;
+
+        balance
+            .parse()
+            .context("Unable to parse the account balance")
+    }
 }
 
 fn blake2<const N: usize>(data: Vec<u8>) -> [u8; N] {

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

@@ -97,6 +97,13 @@ impl DataAvailabilityClient for CelestiaClient {
     fn blob_size_limit(&self) -> Option<usize> {
         Some(1973786) // almost 2MB
     }
+
+    async fn balance(&self) -> Result<u64, DAError> {
+        self.client
+            .balance()
+            .await
+            .map_err(to_non_retriable_da_error)
+    }
 }
 
 impl Debug for CelestiaClient {