feat(eth_sender): Properly calculate blob price (#4406)

## What ❔

<!-- What are the changes this PR brings about? -->
<!-- Example: This PR adds a PR template to the repo. -->
<!-- (For bigger PRs adding more context is appreciated) -->

## Why ❔

<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- The `Why` has to be clear to non-Matter Labs entities running their
own ZK Chain -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## Is this a breaking change?
- [ ] Yes
- [ ] No

## Operational changes
<!-- Any config changes? Any new flags? Any changes to any scripts? -->
<!-- Please add anything that non-Matter Labs entities running their own
ZK Chain may need to know -->

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] 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 `zkstack dev fmt` and `zkstack dev
lint`.

Author: Artemka374

core/Cargo.lock

@@ -0,0 +1 @@
+ALTER TABLE l1_batches DROP COLUMN blobs_amount;

core/lib/dal/.sqlx/query-3db31982ac100a2b2de24f8a42b9f5882d69055a3399f046adea196ae053d6a3.json

@@ -0,0 +1 @@
+ALTER TABLE l1_batches ADD COLUMN blobs_amount BIGINT DEFAULT 0;

…d66dd9fbd20941aec20e2e8c3874b15a953.json …b70c38712f13c80c5b474d8d2227958fb74.jsoncore/lib/dal/.sqlx/query-a0ad14cd53afb6c360e70c47dbd79d66dd9fbd20941aec20e2e8c3874b15a953.json renamed to core/lib/dal/.sqlx/query-9e9b1ff2089d168b1b0819bd6fc3fb70c38712f13c80c5b474d8d2227958fb74.json core/lib/dal/.sqlx/query-a0ad14cd53afb6c360e70c47dbd79d66dd9fbd20941aec20e2e8c3874b15a953.json renamed to core/lib/dal/.sqlx/query-9e9b1ff2089d168b1b0819bd6fc3fb70c38712f13c80c5b474d8d2227958fb74.json

@@ -6,7 +6,7 @@ use std::{
 };
 
 use anyhow::Context as _;
-use bigdecimal::{BigDecimal, FromPrimitive};
+use bigdecimal::{BigDecimal, FromPrimitive, ToPrimitive};
 use sqlx::types::chrono::{DateTime, Utc};
 use zksync_db_connection::{
     connection::Connection,
@@ -1116,6 +1116,7 @@ impl BlocksDal<'_, '_> {
         storage_refunds: &[u32],
         pubdata_costs: &[i32],
         predicted_circuits_by_type: CircuitStatistic, // predicted number of circuits for each circuit type
+        bytes_per_blob: u64,
     ) -> anyhow::Result<()> {
         let initial_bootloader_contents_len = initial_bootloader_contents.len();
         let instrumentation = Instrumented::new("mark_l1_batch_as_sealed")
@@ -1143,6 +1144,12 @@ impl BlocksDal<'_, '_> {
         let storage_refunds: Vec<_> = storage_refunds.iter().copied().map(i64::from).collect();
         let pubdata_costs: Vec<_> = pubdata_costs.iter().copied().map(i64::from).collect();
 
+        let blobs_amount = pubdata_input
+            .clone()
+            .map(|input| input.len() as u64)
+            .unwrap_or(0)
+            .div_ceil(bytes_per_blob);
+
         let query = sqlx::query!(
             r#"
             UPDATE l1_batches
@@ -1163,6 +1170,7 @@ impl BlocksDal<'_, '_> {
                 pubdata_costs = $15,
                 pubdata_input = $16,
                 predicted_circuits_by_type = $17,
+                blobs_amount = $18,
                 updated_at = NOW(),
                 sealed_at = NOW(),
                 is_sealed = TRUE
@@ -1190,6 +1198,7 @@ impl BlocksDal<'_, '_> {
             &pubdata_costs,
             pubdata_input,
             serde_json::to_value(predicted_circuits_by_type).unwrap(),
+            blobs_amount as i64,
         );
         let update_result = instrumentation.with(query).execute(self.storage).await?;
 
@@ -1203,6 +1212,29 @@ impl BlocksDal<'_, '_> {
         Ok(())
     }
 
+    pub async fn get_blobs_amount_for_range(
+        &mut self,
+        lower_bound: L1BatchNumber,
+        upper_bound: L1BatchNumber,
+    ) -> DalResult<u64> {
+        let result: u64 = sqlx::query!(
+            r#"
+            SELECT SUM(blobs_amount) FROM l1_batches WHERE number BETWEEN $1 AND $2
+            "#,
+            i64::from(lower_bound.0),
+            i64::from(upper_bound.0)
+        )
+        .instrument("get_blobs_amount_for_range")
+        .fetch_optional(self.storage)
+        .await?
+        .map(|r| r.sum)
+        .unwrap()
+        .map(|x| x.to_i64().unwrap() as u64)
+        .unwrap_or(0);
+
+        Ok(result)
+    }
+
     pub async fn get_unsealed_l1_batch(&mut self) -> DalResult<Option<UnsealedL1BatchHeader>> {
         Self::get_unsealed_l1_batch_inner(self.storage).await
     }
@@ -3584,7 +3616,7 @@ impl BlocksDal<'_, '_> {
 
     pub async fn insert_mock_l1_batch(&mut self, header: &L1BatchHeader) -> anyhow::Result<()> {
         self.insert_l1_batch(header.to_unsealed_header()).await?;
-        self.mark_l1_batch_as_sealed(header, &[], &[], &[], Default::default())
+        self.mark_l1_batch_as_sealed(header, &[], &[], &[], Default::default(), 1)
             .await
     }
 

core/lib/dal/.sqlx/query-fb416f211acb48e62c17df756ce349d7eebea1ad46df430376aac84fd67a88f1.json

@@ -10,12 +10,13 @@ use zksync_types::{
         AggregatedActionType, L1BatchAggregatedActionType, L2BlockAggregatedActionType,
     },
     eth_sender::{EthTx, EthTxBlobSidecar, EthTxFinalityStatus, TxHistory},
+    server_notification::GatewayMigrationNotification,
     Address, L1BatchNumber, L2BlockNumber, SLChainId, H256, U256,
 };
 
 use crate::{
     models::storage_eth_tx::{BlocksEthSenderStats, StorageEthTx, StorageTxHistory},
-    Core,
+    Core, CoreDal,
 };
 
 #[derive(Debug)]
@@ -291,6 +292,72 @@ impl EthSenderDal<'_, '_> {
         Ok(stats)
     }
 
+    /// This query returns the number and sent_at_block of the latest commited batch.
+    /// Query should be primary used for L1, not for gateway settlement layer.
+    pub async fn get_number_and_sent_at_block_for_latest_commited_batch(
+        &mut self,
+        latest_block_number: u32,
+    ) -> DalResult<(u32, u32)> {
+        // if no commited batch found, return first batch number and latest block number
+        let result = sqlx::query!(
+            r#"
+            SELECT number, eth_txs_history.sent_at_block, eth_txs.is_gateway
+            FROM l1_batches
+            INNER JOIN
+                eth_txs_history
+                ON l1_batches.eth_commit_tx_id = eth_txs_history.eth_tx_id
+            INNER JOIN eth_txs ON eth_txs_history.eth_tx_id = eth_txs.id
+            WHERE eth_txs_history.finality_status = 'finalized'
+            ORDER BY number DESC
+            LIMIT 1
+            "#,
+        )
+        .instrument("get_number_and_sent_at_block_for_latest_commited_batch")
+        .fetch_optional(self.storage)
+        .await?;
+
+        // Here we are returning final values based on the following logic:
+        // - If there is no commited batch, we are returning 1 and latest block number
+        // - If the commited batch is not a gateway batch, we are returning the batch number and the sent at block
+        // - If the commited batch was a gateway batch, we are checking, if there was a gateway migration GW -> L1,
+        // if so, we are returning the block number of the gateway migration, otherwise we are returning the latest block number
+        match result {
+            Some(row) => {
+                let batch_number = row.number as u32;
+                let sent_at_block = if !row.is_gateway {
+                    let block = row.sent_at_block.map(|v| v as u32);
+
+                    if let Some(block) = block {
+                        block
+                    } else {
+                        tracing::error!("There is a commited batch, but no sent at block");
+                        latest_block_number
+                    }
+                } else {
+                    let latest_notification = self
+                        .storage
+                        .server_notifications_dal()
+                        .get_latest_gateway_migration_notification_and_block_number()
+                        .await?;
+
+                    if let Some((
+                        GatewayMigrationNotification::FromGateway,
+                        notification_block_number,
+                    )) = latest_notification
+                    {
+                        notification_block_number.0
+                    } else {
+                        tracing::error!("No latest GW -> L1 migration notification, but trying to calculate blob prices");
+                        latest_block_number
+                    }
+                };
+
+                Ok((batch_number, sent_at_block))
+            }
+            None => Ok((1, latest_block_number)),
+        }
+    }
+
     pub async fn get_eth_tx(&mut self, eth_tx_id: u32) -> sqlx::Result<Option<EthTx>> {
         Ok(sqlx::query_as!(
             StorageEthTx,

core/lib/dal/migrations/20250820140638_count-blob-amount.down.sql

@@ -83,4 +83,24 @@ impl ServerNotificationsDal<'_, '_> {
 
         Ok(notification)
     }
+
+    pub async fn get_latest_gateway_migration_notification_and_block_number(
+        &mut self,
+    ) -> DalResult<Option<(GatewayMigrationNotification, L1BlockNumber)>> {
+        let notification = self
+            .get_last_notification_by_topics(
+                &GatewayMigrationNotification::get_server_notifier_topics(),
+            )
+            .await?;
+
+        let notification = notification.map(|notification| {
+            (
+                GatewayMigrationNotification::from_topic(notification.main_topic)
+                    .expect("Invalid topic"),
+                notification.l1_block_number,
+            )
+        });
+
+        Ok(notification)
+    }
 }

core/lib/dal/migrations/20250820140638_count-blob-amount.up.sql

@@ -271,7 +271,7 @@ mod tests {
             .await
             .unwrap();
         conn.blocks_dal()
-            .mark_l1_batch_as_sealed(&l1_batch_header, &[], &[], &[], Default::default())
+            .mark_l1_batch_as_sealed(&l1_batch_header, &[], &[], &[], Default::default(), 1)
             .await
             .unwrap();
         conn.blocks_dal()