feat(eth-sender): fix for missing eth_txs_history entries (#2236)

This change introduces an invariant "If the transaction may have been
sent, then the sent_at_block is set". This way we never remove
eth_txs_history entries if the transaction may be included in block. The
downside is that we can't distinguish between not sent transaction and
one that has not been mined, but you can't have it both.
The next step is to remove "send_unsent_transactions" step, but it can't
be done in this PR as there might be transactions in db without
sent_at_block set

---------

Signed-off-by: tomg10 <lemures64@gmail.com>

core/lib/dal/src/eth_sender_dal.rs

@@ -221,6 +221,7 @@ impl EthSenderDal<'_, '_> {
         Ok(eth_tx.into())
     }
 
+    #[allow(clippy::too_many_arguments)]
     pub async fn insert_tx_history(
         &mut self,
         eth_tx_id: u32,
@@ -229,6 +230,7 @@ impl EthSenderDal<'_, '_> {
         blob_base_fee_per_gas: Option<u64>,
         tx_hash: H256,
         raw_signed_tx: &[u8],
+        sent_at_block: u32,
     ) -> anyhow::Result<Option<u32>> {
         let priority_fee_per_gas =
             i64::try_from(priority_fee_per_gas).context("Can't convert u64 to i64")?;
@@ -247,10 +249,12 @@ impl EthSenderDal<'_, '_> {
                     signed_raw_tx,
                     created_at,
                     updated_at,
-                    blob_base_fee_per_gas
+                    blob_base_fee_per_gas,
+                    sent_at_block,
+                    sent_at
                 )
             VALUES
-                ($1, $2, $3, $4, $5, NOW(), NOW(), $6)
+                ($1, $2, $3, $4, $5, NOW(), NOW(), $6, $7, NOW())
             ON CONFLICT (tx_hash) DO NOTHING
             RETURNING
                 id
@@ -261,6 +265,7 @@ impl EthSenderDal<'_, '_> {
             tx_hash,
             raw_signed_tx,
             blob_base_fee_per_gas.map(|v| v as i64),
+            sent_at_block as i32
         )
         .fetch_optional(self.storage.conn())
         .await?

core/node/eth_sender/src/eth_tx_manager.rs

@@ -190,12 +190,13 @@ impl EthTxManager {
                 blob_base_fee_per_gas,
                 signed_tx.hash,
                 signed_tx.raw_tx.as_ref(),
+                current_block.0,
             )
             .await
             .unwrap()
         {
             if let Err(error) = self
-                .send_raw_transaction(storage, tx_history_id, signed_tx.raw_tx, current_block)
+                .send_raw_transaction(storage, tx_history_id, signed_tx.raw_tx)
                 .await
             {
                 tracing::warn!(
@@ -216,17 +217,9 @@ impl EthTxManager {
         storage: &mut Connection<'_, Core>,
         tx_history_id: u32,
         raw_tx: RawTransactionBytes,
-        current_block: L1BlockNumber,
     ) -> Result<(), EthSenderError> {
         match self.l1_interface.send_raw_tx(raw_tx).await {
-            Ok(_) => {
-                storage
-                    .eth_sender_dal()
-                    .set_sent_at_block(tx_history_id, current_block.0)
-                    .await
-                    .unwrap();
-                Ok(())
-            }
+            Ok(_) => Ok(()),
             Err(error) => {
                 // In transient errors, server may have received the transaction
                 // we don't want to loose record about it in case that happens
@@ -401,16 +394,22 @@ impl EthTxManager {
 
                 self.apply_tx_status(storage, &eth_tx, tx_status, l1_block_numbers.finalized)
                     .await;
-            } else if let Err(error) = self
-                .send_raw_transaction(
-                    storage,
-                    tx.id,
-                    RawTransactionBytes::new_unchecked(tx.signed_raw_tx.clone()),
-                    l1_block_numbers.latest,
-                )
-                .await
-            {
-                tracing::warn!("Error sending transaction {tx:?}: {error}");
+            } else {
+                storage
+                    .eth_sender_dal()
+                    .set_sent_at_block(tx.id, l1_block_numbers.latest.0)
+                    .await
+                    .unwrap();
+                if let Err(error) = self
+                    .send_raw_transaction(
+                        storage,
+                        tx.id,
+                        RawTransactionBytes::new_unchecked(tx.signed_raw_tx.clone()),
+                    )
+                    .await
+                {
+                    tracing::warn!("Error sending transaction {tx:?}: {error}");
+                }
             }
         }
     }