Fix/nonce to low recovery (#41)

* fix: shutdown on background tasks

* fix: nonce to low if gets out of sync recovery

* fix: remove startup filtering

Author: joshstevens19

crates/cli/Makefile

@@ -16,8 +16,10 @@ start_base:
 	RUST_BACKTRACE=1 cargo run -- start --path $(CURDIR)/../../playground/base
 start_sepolia:
 	RUST_BACKTRACE=1 cargo run -- start --path $(CURDIR)/../../playground/sepolia
-sepolia_ethereum_fireblocks:
+start_sepolia_ethereum_fireblocks:
 	RUST_BACKTRACE=1 cargo run -- start --path $(CURDIR)/../../playground/sepolia_ethereum_fireblocks
+start_e2e:
+	RUST_BACKTRACE=1 cargo run -- start --path $(CURDIR)/../e2e-tests
 
 auth_login:
 	cargo run -- auth login

crates/core/src/background_tasks/automatic_top_up_task.rs

@@ -11,6 +11,7 @@ use crate::{
     relayer::Relayer,
     safe_proxy::SafeProxyManager,
     shared::common_types::EvmAddress,
+    shutdown::subscribe_to_shutdown,
     transaction::queue_system::TransactionsQueues,
     yaml::{AllOrAddresses, Erc20TokenConfig, NativeTokenConfig, NetworkAutomaticTopUpConfig},
     SetupConfig,
@@ -66,6 +67,7 @@ impl AutomaticTopUpTask {
         info!("Starting automatic top-up background task");
 
         self.refresh_relayer_cache().await;
+        let mut shutdown_rx = subscribe_to_shutdown();
 
         loop {
             tokio::select! {
@@ -75,6 +77,10 @@ impl AutomaticTopUpTask {
                 _ = self.top_up_check_interval.tick() => {
                     self.check_and_top_up_addresses().await;
                 }
+                _ = shutdown_rx.recv() => {
+                    info!("Shutdown signal received, stopping automatic top-up task");
+                    break;
+                }
             }
         }
     }

crates/core/src/background_tasks/balance_monitor.rs

@@ -6,7 +6,8 @@ use tracing::{error, info, warn};
 
 use crate::common_types::EvmAddress;
 use crate::{
-    network::ChainId, postgres::PostgresClient, provider::EvmProvider, webhooks::WebhookManager,
+    network::ChainId, postgres::PostgresClient, provider::EvmProvider,
+    shutdown::subscribe_to_shutdown, webhooks::WebhookManager,
 };
 
 fn get_minimum_balance_threshold(chain_id: &ChainId) -> u128 {
@@ -28,19 +29,26 @@ pub async fn balance_monitor(
 
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_secs(600));
+        let mut shutdown_rx = subscribe_to_shutdown();
 
         loop {
-            interval.tick().await;
-
-            info!("Starting balance monitoring check");
+            tokio::select! {
+                _ = interval.tick() => {
+                    info!("Starting balance monitoring check");
+
+                    for provider in providers.iter() {
+                        if let Err(e) = check_balances_for_chain(provider, &db, &webhook_manager).await {
+                            error!("Failed to check balances for chain {}: {}", provider.chain_id, e);
+                        }
+                    }
 
-            for provider in providers.iter() {
-                if let Err(e) = check_balances_for_chain(provider, &db, &webhook_manager).await {
-                    error!("Failed to check balances for chain {}: {}", provider.chain_id, e);
+                    info!("Completed balance monitoring check");
+                }
+                _ = shutdown_rx.recv() => {
+                    info!("Shutdown signal received, stopping balance monitor");
+                    break;
                 }
             }
-
-            info!("Completed balance monitoring check");
         }
     });
 

crates/core/src/background_tasks/webhook_manager_task.rs

@@ -1,4 +1,4 @@
-use crate::{provider::EvmProvider, webhooks::WebhookManager};
+use crate::{provider::EvmProvider, shutdown::subscribe_to_shutdown, webhooks::WebhookManager};
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use tracing::info;
@@ -34,6 +34,8 @@ async fn run_webhook_background_tasks(webhook_manager: Arc<Mutex<WebhookManager>
 
     info!("Starting webhook background processing loops");
 
+    let mut shutdown_rx = subscribe_to_shutdown();
+
     loop {
         tokio::select! {
             _ = retry_interval.tick() => {
@@ -68,6 +70,10 @@ async fn run_webhook_background_tasks(webhook_manager: Arc<Mutex<WebhookManager>
             _ = database_cleanup_interval.tick() => {
                 cleanup_webhook_database_history(webhook_manager.clone()).await;
             }
+            _ = shutdown_rx.recv() => {
+                info!("Shutdown signal received, stopping webhook manager");
+                break;
+            }
         }
     }
 }

crates/core/src/gas/blob_gas_oracle.rs

@@ -1,6 +1,9 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::{network::ChainId, provider::EvmProvider, transaction::types::TransactionSpeed};
+use crate::{
+    network::ChainId, provider::EvmProvider, shutdown::subscribe_to_shutdown,
+    transaction::types::TransactionSpeed,
+};
 use serde::{Deserialize, Serialize};
 use tokio::{
     sync::Mutex,
@@ -130,16 +133,24 @@ pub async fn blob_gas_oracle(
 
         tokio::spawn(async move {
             let mut interval = time::interval(Duration::from_secs(5));
+            let mut shutdown_rx = subscribe_to_shutdown();
+
             loop {
-                interval.tick().await;
-
-                let blob_gas_price_result = provider.calculate_ethereum_blob_gas_price().await;
-                if let Ok(blob_gas_price) = blob_gas_price_result {
-                    cache
-                        .lock()
-                        .await
-                        .update_blob_gas_price(provider.chain_id, blob_gas_price)
-                        .await;
+                tokio::select! {
+                    _ = interval.tick() => {
+                        let blob_gas_price_result = provider.calculate_ethereum_blob_gas_price().await;
+                        if let Ok(blob_gas_price) = blob_gas_price_result {
+                            cache
+                                .lock()
+                                .await
+                                .update_blob_gas_price(provider.chain_id, blob_gas_price)
+                                .await;
+                        }
+                    }
+                    _ = shutdown_rx.recv() => {
+                        info!("Shutdown signal received, stopping blob gas oracle for provider: {}", provider.name);
+                        break;
+                    }
                 }
             }
         });

crates/core/src/gas/gas_oracle.rs

@@ -1,7 +1,10 @@
 use std::{collections::HashMap, sync::Arc};
 
 use super::fee_estimator::{GasEstimatorResult, GasPriceResult};
-use crate::{network::ChainId, provider::EvmProvider, transaction::types::TransactionSpeed};
+use crate::{
+    network::ChainId, provider::EvmProvider, shutdown::subscribe_to_shutdown,
+    transaction::types::TransactionSpeed,
+};
 use tokio::{
     sync::Mutex,
     time::{self, Duration},
@@ -91,16 +94,24 @@ pub async fn gas_oracle(
 
         tokio::spawn(async move {
             let mut interval = time::interval(Duration::from_secs(5));
-            loop {
-                interval.tick().await;
+            let mut shutdown_rx = subscribe_to_shutdown();
 
-                let gas_price_result = provider.calculate_gas_price().await;
-                match gas_price_result {
-                    Ok(gas_price) => {
-                        cache.lock().await.update_gas_price(provider.chain_id, gas_price).await;
+            loop {
+                tokio::select! {
+                    _ = interval.tick() => {
+                        let gas_price_result = provider.calculate_gas_price().await;
+                        match gas_price_result {
+                            Ok(gas_price) => {
+                                cache.lock().await.update_gas_price(provider.chain_id, gas_price).await;
+                            }
+                            Err(err) => {
+                                error!("Failed to get gas price for provider: {} - error {} - try again in 10s", provider.name, err);
+                            }
+                        }
                     }
-                    Err(err) => {
-                        error!("Failed to get gas price for provider: {} - error {} - try again in 10s", provider.name, err);
+                    _ = shutdown_rx.recv() => {
+                        info!("Shutdown signal received, stopping gas oracle for provider: {}", provider.name);
+                        break;
                     }
                 }
             }

crates/core/src/transaction/db/write.rs

@@ -8,8 +8,8 @@ use crate::{
         utils::option_if,
     },
     transaction::types::{
-        Transaction, TransactionData, TransactionHash, TransactionId, TransactionStatus,
-        TransactionValue,
+        Transaction, TransactionData, TransactionHash, TransactionId, TransactionNonce,
+        TransactionStatus, TransactionValue,
     },
 };
 use alloy::network::AnyTransactionReceipt;
@@ -427,6 +427,47 @@ impl PostgresClient {
         Ok(())
     }
 
+    pub async fn transaction_update_nonce(
+        &mut self,
+        transaction_id: &TransactionId,
+        nonce: &TransactionNonce,
+    ) -> Result<(), PostgresError> {
+        let mut conn = self.pool.get().await?;
+        let trans = conn.transaction().await.map_err(PostgresError::PgError)?;
+
+        trans
+            .execute(
+                "UPDATE relayer.transaction SET nonce = $2 WHERE id = $1",
+                &[&transaction_id, &(nonce.into_inner() as i64)],
+            )
+            .await?;
+
+        trans
+            .execute(
+                "
+                    INSERT INTO relayer.transaction_audit_log (
+                        id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, gas_limit,
+                        speed, status, expires_at, queued_at, sent_at, mined_at, confirmed_at,
+                        failed_at, failed_reason, hash, sent_max_priority_fee_per_gas,
+                        sent_max_fee_per_gas, gas_price, block_hash, block_number, external_id
+                    )
+                    SELECT 
+                        id, relayer_id, \"to\", \"from\", $2, chain_id, data, value, blobs, gas_limit,
+                        speed, status, expires_at, queued_at, sent_at, mined_at, confirmed_at,
+                        failed_at, failed_reason, hash, sent_max_priority_fee_per_gas,
+                        sent_max_fee_per_gas, gas_price, block_hash, block_number, external_id
+                    FROM relayer.transaction
+                    WHERE id = $1;
+                ",
+                &[&transaction_id, &(nonce.into_inner() as i64)],
+            )
+            .await?;
+
+        trans.commit().await?;
+
+        Ok(())
+    }
+
     pub async fn transaction_expired(
         &mut self,
         transaction_id: &TransactionId,

crates/core/src/transaction/queue_system/transactions_queue.rs

@@ -1398,4 +1398,30 @@ impl TransactionsQueue {
         let address = self.relay_address();
         self.evm_provider.get_balance(&address).await
     }
+
+    pub async fn update_pending_transaction_nonce(
+        &self,
+        transaction_id: &TransactionId,
+        new_nonce: TransactionNonce,
+    ) {
+        let mut pending = self.pending_transactions.lock().await;
+        if let Some(transaction) = pending.iter_mut().find(|tx| tx.id == *transaction_id) {
+            transaction.nonce = new_nonce;
+        }
+    }
+
+    pub async fn update_inmempool_transaction_nonce(
+        &self,
+        transaction_id: &TransactionId,
+        new_nonce: TransactionNonce,
+    ) {
+        let mut inmempool = self.inmempool_transactions.lock().await;
+        if let Some(competitive_tx) =
+            inmempool.iter_mut().find(|ctx| ctx.get_transaction_by_id(transaction_id).is_some())
+        {
+            if let Some(transaction) = competitive_tx.get_transaction_by_id_mut(transaction_id) {
+                transaction.nonce = new_nonce;
+            }
+        }
+    }
 }