feat(eth-sender): gateway support for eth tx manager (#2593)

Signed-off-by: tomg10 <[email protected]>
Co-authored-by: Stanislav Breadless <[email protected]>

core/lib/config/src/configs/eth_sender.rs

@@ -24,7 +24,7 @@ impl EthConfig {
         Self {
             sender: Some(SenderConfig {
                 aggregated_proof_sizes: vec![1],
-                wait_confirmations: Some(1),
+                wait_confirmations: Some(10),
                 tx_poll_period: 1,
                 aggregate_tx_poll_period: 1,
                 max_txs_in_flight: 30,

core/lib/dal/migrations/20240803083814_add_is_gateway_column_to_eth_txs.down.sql

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

core/lib/dal/migrations/20240803083814_add_is_gateway_column_to_eth_txs.up.sql

@@ -0,0 +1 @@
+ALTER TABLE eth_txs ADD COLUMN is_gateway BOOLEAN NOT NULL DEFAULT FALSE;

core/lib/dal/src/blocks_dal.rs

@@ -2548,7 +2548,16 @@ mod tests {
 
     async fn save_mock_eth_tx(action_type: AggregatedActionType, conn: &mut Connection<'_, Core>) {
         conn.eth_sender_dal()
-            .save_eth_tx(1, vec![], action_type, Address::default(), 1, None, None)
+            .save_eth_tx(
+                1,
+                vec![],
+                action_type,
+                Address::default(),
+                1,
+                None,
+                None,
+                false,
+            )
             .await
             .unwrap();
     }

core/lib/dal/src/blocks_web3_dal.rs

@@ -977,6 +977,7 @@ mod tests {
                 0,
                 None,
                 None,
+                false,
             )
             .await
             .unwrap();

core/lib/dal/src/eth_sender_dal.rs

@@ -25,6 +25,7 @@ impl EthSenderDal<'_, '_> {
     pub async fn get_inflight_txs(
         &mut self,
         operator_address: Option<Address>,
+        is_gateway: bool,
     ) -> sqlx::Result<Vec<EthTx>> {
         let txs = sqlx::query_as!(
             StorageEthTx,
@@ -36,6 +37,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 is_gateway = $2
                 AND id <= (
                     SELECT
                         COALESCE(MAX(eth_tx_id), 0)
@@ -45,17 +47,40 @@ impl EthSenderDal<'_, '_> {
                     WHERE
                         eth_txs_history.sent_at_block IS NOT NULL
                         AND eth_txs.from_addr IS NOT DISTINCT FROM $1
+                        AND is_gateway = $2
                 )
             ORDER BY
                 id
             "#,
             operator_address.as_ref().map(|h160| h160.as_bytes()),
+            is_gateway
         )
         .fetch_all(self.storage.conn())
         .await?;
         Ok(txs.into_iter().map(|tx| tx.into()).collect())
     }
 
+    pub async fn get_non_gateway_inflight_txs_count_for_gateway_migration(
+        &mut self,
+    ) -> sqlx::Result<usize> {
+        let count = sqlx::query!(
+            r#"
+            SELECT
+                COUNT(*)
+            FROM
+                eth_txs
+            WHERE
+                confirmed_eth_tx_history_id IS NULL
+                AND is_gateway = FALSE
+            "#
+        )
+        .fetch_one(self.storage.conn())
+        .await?
+        .count
+        .unwrap();
+        Ok(count.try_into().unwrap())
+    }
+
     pub async fn get_eth_l1_batches(&mut self) -> sqlx::Result<L1BatchEthSenderStats> {
         struct EthTxRow {
             number: i64,
@@ -132,6 +157,7 @@ impl EthSenderDal<'_, '_> {
         &mut self,
         limit: u64,
         operator_address: &Option<Address>,
+        is_gateway: bool,
     ) -> sqlx::Result<Vec<EthTx>> {
         let txs = sqlx::query_as!(
             StorageEthTx,
@@ -142,6 +168,7 @@ impl EthSenderDal<'_, '_> {
                 eth_txs
             WHERE
                 from_addr IS NOT DISTINCT FROM $2 -- can't just use equality as NULL != NULL
+                AND is_gateway = $3
                 AND id > (
                     SELECT
                         COALESCE(MAX(eth_tx_id), 0)
@@ -151,6 +178,7 @@ impl EthSenderDal<'_, '_> {
                     WHERE
                         eth_txs_history.sent_at_block IS NOT NULL
                         AND eth_txs.from_addr IS NOT DISTINCT FROM $2
+                        AND is_gateway = $3
                 )
             ORDER BY
                 id
@@ -159,6 +187,7 @@ impl EthSenderDal<'_, '_> {
             "#,
             limit as i64,
             operator_address.as_ref().map(|h160| h160.as_bytes()),
+            is_gateway
         )
         .fetch_all(self.storage.conn())
         .await?;
@@ -202,6 +231,7 @@ impl EthSenderDal<'_, '_> {
         predicted_gas_cost: u32,
         from_address: Option<Address>,
         blob_sidecar: Option<EthTxBlobSidecar>,
+        is_gateway: bool,
     ) -> sqlx::Result<EthTx> {
         let address = format!("{:#x}", contract_address);
         let eth_tx = sqlx::query_as!(
@@ -217,10 +247,11 @@ impl EthSenderDal<'_, '_> {
                     created_at,
                     updated_at,
                     from_addr,
-                    blob_sidecar
+                    blob_sidecar,
+                    is_gateway
                 )
             VALUES
-                ($1, $2, $3, $4, $5, NOW(), NOW(), $6, $7)
+                ($1, $2, $3, $4, $5, NOW(), NOW(), $6, $7, $8)
             RETURNING
                 *
             "#,
@@ -232,6 +263,7 @@ impl EthSenderDal<'_, '_> {
             from_address.as_ref().map(Address::as_bytes),
             blob_sidecar.map(|sidecar| bincode::serialize(&sidecar)
                 .expect("can always bincode serialize EthTxBlobSidecar; qed")),
+            is_gateway,
         )
         .fetch_one(self.storage.conn())
         .await?;

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

@@ -29,6 +29,7 @@ pub struct StorageEthTx {
     //
     // Format a `bincode`-encoded `EthTxBlobSidecar` enum.
     pub blob_sidecar: Option<Vec<u8>>,
+    pub is_gateway: bool,
 }
 
 #[derive(Debug, Default)]
@@ -83,6 +84,7 @@ impl From<StorageEthTx> for EthTx {
             blob_sidecar: tx.blob_sidecar.map(|b| {
                 bincode::deserialize(&b).expect("EthTxBlobSidecar is encoded correctly; qed")
             }),
+            is_gateway: tx.is_gateway,
         }
     }
 }

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

@@ -107,7 +107,7 @@ impl MockSettlementLayerInner {
         self.block_number += confirmations;
         let nonce = self.current_nonce;
         self.current_nonce += 1;
-        tracing::info!("Executing tx with hash {tx_hash:?}, success: {success}, current nonce: {}, confirmations: {confirmations}", self.current_nonce);
+        tracing::info!("Executing tx with hash {tx_hash:?} at block {}, success: {success}, current nonce: {}, confirmations: {confirmations}", self.block_number - confirmations, self.current_nonce);
         let tx_nonce = self.sent_txs[&tx_hash].nonce;
 
         if non_ordering_confirmations {

core/lib/types/src/eth_sender.rs

@@ -51,6 +51,7 @@ pub struct EthTx {
     /// this transaction. If it is set to `None` this transaction was sent by the main operator.
     pub from_addr: Option<Address>,
     pub blob_sidecar: Option<EthTxBlobSidecar>,
+    pub is_gateway: bool,
 }
 
 impl std::fmt::Debug for EthTx {