added tests

core/lib/dal/src/eth_sender_dal.rs

@@ -1,6 +1,6 @@
 use std::{convert::TryFrom, str::FromStr};
 
-use anyhow::Context as _;
+use anyhow::{anyhow, Context as _};
 use sqlx::types::chrono::{DateTime, Utc};
 use zksync_db_connection::{connection::Connection, interpolate_query, match_query_as};
 use zksync_types::{
@@ -35,6 +35,7 @@ impl EthSenderDal<'_, '_> {
             WHERE
                 from_addr IS NOT DISTINCT FROM $1 -- can't just use equality as NULL != NULL
                 AND confirmed_eth_tx_history_id IS NULL
+                AND has_failed = FALSE
                 AND is_gateway = $2
                 AND id <= (
                     SELECT
@@ -69,6 +70,7 @@ impl EthSenderDal<'_, '_> {
                 eth_txs
             WHERE
                 confirmed_eth_tx_history_id IS NULL
+                AND has_failed = FALSE
                 AND is_gateway = FALSE
             "#
         )
@@ -634,7 +636,13 @@ impl EthSenderDal<'_, '_> {
         .context("count field is missing")
     }
 
-    pub async fn clear_failed_transactions(&mut self) -> sqlx::Result<()> {
+    pub async fn clear_failed_transactions(&mut self) -> anyhow::Result<()> {
+        if self.count_all_inflight_txs().await.unwrap() != 0 {
+            return Err(anyhow!(
+                "There are still some in-flight txs, cannot proceed. \
+            Please wait for eth-sender to process all in-flight txs and try again!"
+            ));
+        }
         sqlx::query!(
             r#"
             DELETE FROM eth_txs

core/lib/eth_client/src/clients/mock.rs

@@ -137,6 +137,7 @@ impl MockSettlementLayerInner {
             unimplemented!("Getting nonce for custom account is not supported");
         }
 
+        tracing::info!("Getting nonce for account {address:?} at block {block:?}, pending nonce: {}, current nonce: {}", self.pending_nonce, self.current_nonce);
         match block {
             web3::BlockNumber::Number(block_number) => {
                 let mut nonce_range = self.nonces.range(..=block_number.as_u64());

core/node/block_reverter/src/lib.rs

@@ -606,21 +606,6 @@ impl BlockReverter {
     /// Clears failed L1 transactions.
     pub async fn clear_failed_l1_transactions(&self) -> anyhow::Result<()> {
         tracing::info!("Clearing failed L1 transactions");
-        if self
-            .connection_pool
-            .connection()
-            .await?
-            .eth_sender_dal()
-            .count_all_inflight_txs()
-            .await
-            .unwrap()
-            != 0
-        {
-            tracing::error!(
-                "There are still some in-flight txs, cannot proceed. \
-            Please wait for eth-sender to process all in-flight txs and try again!"
-            );
-        }
         self.connection_pool
             .connection()
             .await?

core/node/eth_sender/src/eth_tx_aggregator.rs

@@ -21,7 +21,7 @@ use zksync_types::{
     protocol_version::{L1VerifierConfig, PACKED_SEMVER_MINOR_MASK},
     pubdata_da::PubdataDA,
     settlement::SettlementMode,
-    web3::{contract::Error as Web3ContractError, BlockNumber},
+    web3::contract::Error as Web3ContractError,
     Address, L2ChainId, ProtocolVersionId, SLChainId, H256, U256,
 };
 
@@ -49,17 +49,13 @@ pub struct MulticallData {
 pub struct EthTxAggregator {
     aggregator: Aggregator,
     eth_client: Box<dyn BoundEthInterface>,
+    eth_client_blobs: Option<Box<dyn BoundEthInterface>>,
     config: SenderConfig,
     timelock_contract_address: Address,
     l1_multicall3_address: Address,
     pub(super) state_transition_chain_contract: Address,
     functions: ZkSyncFunctions,
     rollup_chain_id: L2ChainId,
-    /// If set to `Some` node is operating in the 4844 mode with two operator
-    /// addresses at play: the main one and the custom address for sending commit
-    /// transactions. The `Some` then contains the address of this custom operator
-    /// address.
-    custom_commit_sender_addr: Option<Address>,
     pool: ConnectionPool<Core>,
     settlement_mode: SettlementMode,
     sl_chain_id: SLChainId,
@@ -77,11 +73,11 @@ impl EthTxAggregator {
         config: SenderConfig,
         aggregator: Aggregator,
         eth_client: Box<dyn BoundEthInterface>,
+        eth_client_blobs: Option<Box<dyn BoundEthInterface>>,
         timelock_contract_address: Address,
         l1_multicall3_address: Address,
         state_transition_chain_contract: Address,
         rollup_chain_id: L2ChainId,
-        custom_commit_sender_addr: Option<Address>,
         settlement_mode: SettlementMode,
     ) -> Self {
         let eth_client = eth_client.for_component("eth_tx_aggregator");
@@ -93,12 +89,12 @@ impl EthTxAggregator {
             config,
             aggregator,
             eth_client,
+            eth_client_blobs,
             timelock_contract_address,
             l1_multicall3_address,
             state_transition_chain_contract,
             functions,
             rollup_chain_id,
-            custom_commit_sender_addr,
             pool,
             settlement_mode,
             sl_chain_id,
@@ -598,11 +594,14 @@ impl EthTxAggregator {
     ) -> Result<EthTx, EthSenderError> {
         let mut transaction = storage.start_transaction().await.unwrap();
         let op_type = aggregated_op.get_action_type();
-        // We may be using a custom sender for commit transactions, so use this
+        // We may be using a custom sender for commit tester.assert_inflight_txs_count_equals(0).await;transactions, so use this
         // var whatever it actually is: a `None` for single-addr operator or `Some`
         // for multi-addr operator in 4844 mode.
         let sender_addr = match (op_type, is_gateway) {
-            (AggregatedActionType::Commit, false) => self.custom_commit_sender_addr,
+            (AggregatedActionType::Commit, false) => self
+                .eth_client_blobs
+                .as_deref()
+                .map(BoundEthInterface::sender_account),
             (_, _) => None,
         };
         let nonce = self.get_next_nonce(&mut transaction, sender_addr).await?;
@@ -659,23 +658,14 @@ impl EthTxAggregator {
             .await
             .unwrap()
             .unwrap_or(0);
+        let client = if from_addr.is_some() {
+            self.eth_client_blobs.as_deref().unwrap()
+        } else {
+            self.eth_client.as_ref()
+        };
         // We can execute some txs using operator account or remove some txs from the database
         // We have to consider this fact and get the max nonce.
-        Ok(if from_addr.is_none() {
-            let base_nonce = self.eth_client.pending_nonce().await.unwrap().as_u64();
-            db_nonce.max(base_nonce)
-        } else {
-            let base_nonce_custom_commit_sender = (*self.eth_client)
-                .as_ref()
-                .nonce_at_for_account(
-                    self.custom_commit_sender_addr
-                        .expect("custom_commit_sender_addr should not be empty"),
-                    BlockNumber::Pending,
-                )
-                .await
-                .unwrap()
-                .as_u64();
-            db_nonce.max(base_nonce_custom_commit_sender)
-        })
+        let base_nonce = client.pending_nonce().await.unwrap().as_u64();
+        Ok(db_nonce.max(base_nonce))
     }
 }

core/node/eth_sender/src/tester.rs

@@ -5,7 +5,7 @@ use zksync_config::{
     ContractsConfig, EthConfig, GasAdjusterConfig,
 };
 use zksync_dal::{Connection, ConnectionPool, Core, CoreDal};
-use zksync_eth_client::{clients::MockSettlementLayer, BaseFees, BoundEthInterface};
+use zksync_eth_client::{clients::MockSettlementLayer, BaseFees};
 use zksync_l1_contract_interface::i_executor::methods::{ExecuteBatches, ProveBatches};
 use zksync_node_fee_model::l1_gas_price::{GasAdjuster, GasAdjusterClient};
 use zksync_node_test_utils::{create_l1_batch, l1_batch_metadata_to_commitment_artifacts};
@@ -119,7 +119,7 @@ pub(crate) struct EthSenderTester {
     pub gas_adjuster: Arc<GasAdjuster>,
     pub pubdata_sending_mode: PubdataSendingMode,
     next_l1_batch_number_to_seal: L1BatchNumber,
-    next_l1_batch_number_to_commit: L1BatchNumber,
+    pub next_l1_batch_number_to_commit: L1BatchNumber,
     next_l1_batch_number_to_prove: L1BatchNumber,
     next_l1_batch_number_to_execute: L1BatchNumber,
     tx_sent_in_last_iteration_count: usize,
@@ -238,13 +238,6 @@ impl EthSenderTester {
 
         let eth_sender = eth_sender_config.sender.clone().unwrap();
 
-        let custom_commit_sender_addr =
-            if aggregator_operate_4844_mode && commitment_mode == L1BatchCommitmentMode::Rollup {
-                Some(gateway_blobs.sender_account())
-            } else {
-                None
-            };
-
         let aggregator = EthTxAggregator::new(
             connection_pool.clone(),
             SenderConfig {
@@ -260,12 +253,16 @@ impl EthSenderTester {
                 commitment_mode,
             ),
             gateway.clone(),
+            if aggregator_operate_4844_mode && commitment_mode == L1BatchCommitmentMode::Rollup {
+                Some(gateway_blobs.clone())
+            } else {
+                None
+            },
             // ZKsync contract address
             Address::random(),
             contracts_config.l1_multicall3_addr,
             STATE_TRANSITION_CONTRACT_ADDRESS,
             Default::default(),
-            custom_commit_sender_addr,
             SettlementMode::SettlesToL1,
         )
         .await;
@@ -595,6 +592,14 @@ impl EthSenderTester {
         );
     }
 
+    pub async fn clear_failed_txs_failed_attempts(&mut self) -> anyhow::Result<()> {
+        self.storage()
+            .await
+            .eth_sender_dal()
+            .clear_failed_transactions()
+            .await
+    }
+
     pub async fn assert_inflight_txs_count_equals(&mut self, value: usize) {
         let inflight_count = if !self.is_l2 {
             //sanity check
@@ -626,7 +631,8 @@ impl EthSenderTester {
 
         assert_eq!(
             inflight_count, value,
-            "Unexpected number of in-flight transactions"
+            "Unexpected number of in-flight transactions, expected: {}, got: {}",
+            value, inflight_count
         );
     }
 }