feat(eth-sender): added chain_id column to eth_txs + support for gate…

…way in tx_aggregator (#2685)

Signed-off-by: tomg10 <[email protected]>

core/bin/zksync_server/src/node_builder.rs

@@ -434,6 +434,11 @@ impl MainNodeBuilder {
             self.contracts_config.clone(),
             self.genesis_config.l2_chain_id,
             self.genesis_config.l1_batch_commit_data_generator_mode,
+            self.configs
+                .eth
+                .as_ref()
+                .and_then(|x| Some(x.gas_adjuster?.settlement_mode))
+                .unwrap_or(SettlementMode::SettlesToL1),
         ));
 
         Ok(self)

core/lib/dal/migrations/20240819150019_add_chain_id_to_eth_txs.down.sql

@@ -0,0 +1 @@
+ALTER TABLE eth_txs DROP COLUMN chain_id;

core/lib/dal/migrations/20240819150019_add_chain_id_to_eth_txs.up.sql

@@ -0,0 +1 @@
+ALTER TABLE eth_txs ADD COLUMN chain_id BIGINT;

core/lib/dal/src/eth_sender_dal.rs

@@ -404,6 +404,23 @@ impl EthSenderDal<'_, '_> {
         Ok(())
     }
 
+    pub async fn set_chain_id(&mut self, eth_tx_id: u32, chain_id: u64) -> anyhow::Result<()> {
+        sqlx::query!(
+            r#"
+            UPDATE eth_txs
+            SET
+                chain_id = $1
+            WHERE
+                id = $2
+            "#,
+            eth_tx_id as i32,
+            chain_id as i64,
+        )
+        .execute(self.storage.conn())
+        .await?;
+        Ok(())
+    }
+
     pub async fn get_confirmed_tx_hash_by_eth_tx_id(
         &mut self,
         eth_tx_id: u32,
@@ -610,29 +627,29 @@ impl EthSenderDal<'_, '_> {
     pub async fn get_next_nonce(
         &mut self,
         from_address: Option<Address>,
+        is_gateway: bool,
     ) -> sqlx::Result<Option<u64>> {
-        struct NonceRow {
-            nonce: i64,
-        }
-
-        let query = match_query_as!(
-            NonceRow,
-            [
-                "SELECT nonce FROM eth_txs WHERE ",
-                _, // WHERE condition
-                " ORDER BY id DESC LIMIT 1"
-            ],
-            match (from_address) {
-                Some(address) => ("from_addr = $1::bytea"; address.as_bytes()),
-                None => ("from_addr IS NULL";),
-            }
-        );
+        let nonce = sqlx::query!(
+            r#"
+            SELECT
+                nonce
+            FROM
+                eth_txs
+            WHERE
+                from_addr IS NOT DISTINCT FROM $1 -- can't just use equality as NULL != NULL\
+                AND is_gateway = $2
+            ORDER BY
+                id DESC
+            LIMIT
+                1
+            "#,
+            from_address.as_ref().map(|h160| h160.as_bytes()),
+            is_gateway
+        )
+        .fetch_optional(self.storage.conn())
+        .await?;
 
-        let nonce = query
-            .fetch_optional(self.storage.conn())
-            .await?
-            .map(|row| row.nonce as u64);
-        Ok(nonce.map(|n| n + 1))
+        Ok(nonce.map(|row| row.nonce as u64 + 1))
     }
 
     pub async fn mark_failed_transaction(&mut self, eth_tx_id: u32) -> sqlx::Result<()> {

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

@@ -4,7 +4,7 @@ use sqlx::types::chrono::NaiveDateTime;
 use zksync_types::{
     aggregated_operations::AggregatedActionType,
     eth_sender::{EthTx, TxHistory, TxHistoryToSend},
-    Address, L1BatchNumber, Nonce, H256,
+    Address, L1BatchNumber, Nonce, SLChainId, H256,
 };
 
 #[derive(Debug, Clone)]
@@ -30,6 +30,7 @@ pub struct StorageEthTx {
     // Format a `bincode`-encoded `EthTxBlobSidecar` enum.
     pub blob_sidecar: Option<Vec<u8>>,
     pub is_gateway: bool,
+    pub chain_id: Option<i64>,
 }
 
 #[derive(Debug, Default)]
@@ -85,6 +86,9 @@ impl From<StorageEthTx> for EthTx {
                 bincode::deserialize(&b).expect("EthTxBlobSidecar is encoded correctly; qed")
             }),
             is_gateway: tx.is_gateway,
+            chain_id: tx
+                .chain_id
+                .map(|chain_id| SLChainId(chain_id.try_into().unwrap())),
         }
     }
 }

core/lib/types/src/eth_sender.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Serialize};
+use zksync_basic_types::SLChainId;
 
 use crate::{aggregated_operations::AggregatedActionType, Address, Nonce, H256};
 
@@ -52,6 +53,7 @@ pub struct EthTx {
     pub from_addr: Option<Address>,
     pub blob_sidecar: Option<EthTxBlobSidecar>,
     pub is_gateway: bool,
+    pub chain_id: Option<SLChainId>,
 }
 
 impl std::fmt::Debug for EthTx {
@@ -64,6 +66,7 @@ impl std::fmt::Debug for EthTx {
             .field("tx_type", &self.tx_type)
             .field("created_at_timestamp", &self.created_at_timestamp)
             .field("predicted_gas_cost", &self.predicted_gas_cost)
+            .field("chain_id", &self.chain_id)
             .finish()
     }
 }

core/node/eth_sender/src/eth_tx_aggregator.rs

@@ -20,8 +20,9 @@ use zksync_types::{
     l2_to_l1_log::UserL2ToL1Log,
     protocol_version::{L1VerifierConfig, PACKED_SEMVER_MINOR_MASK},
     pubdata_da::PubdataDA,
+    settlement::SettlementMode,
     web3::{contract::Error as Web3ContractError, BlockNumber},
-    Address, L2ChainId, ProtocolVersionId, H256, U256,
+    Address, L2ChainId, ProtocolVersionId, SLChainId, H256, U256,
 };
 
 use super::aggregated_operations::AggregatedOperation;
@@ -62,6 +63,8 @@ pub struct EthTxAggregator {
     /// address.
     custom_commit_sender_addr: Option<Address>,
     pool: ConnectionPool<Core>,
+    settlement_mode: SettlementMode,
+    sl_chain_id: SLChainId,
 }
 
 struct TxData {
@@ -81,6 +84,7 @@ impl EthTxAggregator {
         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");
         let functions = ZkSyncFunctions::default();
@@ -97,6 +101,9 @@ impl EthTxAggregator {
             ),
             None => None,
         };
+
+        let sl_chain_id = (*eth_client).as_ref().fetch_chain_id().await.unwrap();
+
         Self {
             config,
             aggregator,
@@ -110,6 +117,8 @@ impl EthTxAggregator {
             rollup_chain_id,
             custom_commit_sender_addr,
             pool,
+            settlement_mode,
+            sl_chain_id,
         }
     }
 
@@ -578,6 +587,12 @@ impl EthTxAggregator {
             .await
             .unwrap();
 
+        transaction
+            .eth_sender_dal()
+            .set_chain_id(eth_tx.id, self.sl_chain_id.0)
+            .await
+            .unwrap();
+
         transaction
             .blocks_dal()
             .set_eth_tx_id(l1_batch_number_range, eth_tx.id, op_type)
@@ -592,9 +607,10 @@ impl EthTxAggregator {
         storage: &mut Connection<'_, Core>,
         from_addr: Option<Address>,
     ) -> Result<u64, EthSenderError> {
+        let is_gateway = self.settlement_mode.is_gateway();
         let db_nonce = storage
             .eth_sender_dal()
-            .get_next_nonce(from_addr)
+            .get_next_nonce(from_addr, is_gateway)
             .await
             .unwrap()
             .unwrap_or(0);