redis memory leak fix & stronger EOA healing (#48)

* add redis cleanup script

* clean up EOA transaction data for successful and failed transactions

* stronger self-healing and recovery for blocked EOA

* review fixes

* more review fixes

Author: d4mr

Cargo.lock

@@ -27,3 +27,4 @@ futures = "0.3.31"
 moka = { version = "0.12.10", features = ["future"] }
 prometheus = "0.13.4"
 lazy_static = "1.5.0"
+tokio-retry2 = {version = "0.6.0", features = ["jitter"]}

executors/Cargo.toml

@@ -38,6 +38,9 @@ pub enum EoaExecutionError {
         message: String,
         inner_error: Option<EngineError>,
     },
+
+    /// Thirdweb support error that requires retry in place
+    ThirdwebSupportError { message: String },
 }
 
 /// Recovery strategy for an EOA execution error
@@ -115,6 +118,10 @@ impl EoaErrorMapper {
                     EoaExecutionError::AccountError {
                         message: message.to_string(),
                     }
+                } else if msg_lower.contains("we are not able to process your request at this time") {
+                    EoaExecutionError::ThirdwebSupportError {
+                        message: message.to_string(),
+                    }
                 } else {
                     // Not an actionable error - let engine error handle it
                     EoaExecutionError::RpcError {
@@ -204,6 +211,14 @@ impl EoaErrorMapper {
                 retry_delay: None,
             },
 
+            EoaExecutionError::ThirdwebSupportError { .. } => RecoveryStrategy {
+                queue_confirmation: false,
+                recycle_nonce: false,
+                needs_resync: false,
+                retryable: true,
+                retry_delay: Some(Duration::from_secs(1)), // Short delay for retry in place
+            },
+
             EoaExecutionError::RpcError { .. } => {
                 // This should not be used - let engine error handle it
                 RecoveryStrategy {
@@ -231,6 +246,7 @@ impl EoaExecutionError {
             | EoaExecutionError::GasError { message }
             | EoaExecutionError::PoolLimitExceeded { message }
             | EoaExecutionError::AccountError { message }
+            | EoaExecutionError::ThirdwebSupportError { message }
             | EoaExecutionError::RpcError { message, .. } => message,
         }
     }

executors/src/eoa/error_classifier.rs

@@ -503,7 +503,8 @@ impl AtomicEoaExecutorStore {
             // Add new hash:id to submitted (keeping old ones)
             pipeline.zadd(&submitted_key, &submitted_transaction_string, nonce);
 
-            // Still maintain separate hash-to-ID mapping for backward compatibility
+            // Create hash-to-ID mapping for the new gas bump hash
+            // Keep ALL hashes until transaction is fully cleaned up for re-org protection
             pipeline.set(&hash_to_id_key, &submitted_transaction.transaction_id);
 
             // Simply push the new attempt to the attempts list
@@ -554,6 +555,7 @@ impl AtomicEoaExecutorStore {
         pending_transaction: &PendingTransaction,
         error: EoaExecutorWorkerError,
         webhook_queue: Arc<twmq::Queue<WebhookJobHandler>>,
+        failed_transaction_expiry_seconds: u64,
     ) -> Result<(), TransactionStoreError> {
         self.with_lock_check(|pipeline| {
             let pending_key = self.pending_transactions_zset_name();
@@ -563,11 +565,23 @@ impl AtomicEoaExecutorStore {
             // Remove from pending state
             pipeline.zrem(&pending_key, &pending_transaction.transaction_id);
 
-            // Update transaction data with failure
+            // Update transaction data with failure and set expiry
             pipeline.hset(&tx_data_key, "completed_at", now);
             pipeline.hset(&tx_data_key, "failure_reason", error.to_string());
             pipeline.hset(&tx_data_key, "status", "failed");
 
+            // Set expiry on all related keys for failed transactions
+            let transaction_keys =
+                self.get_all_transaction_keys(&pending_transaction.transaction_id);
+
+            let ttl: i64 = i64::try_from(failed_transaction_expiry_seconds)
+                .unwrap_or(i64::MAX)
+                .max(1);
+
+            for key in transaction_keys {
+                pipeline.expire(&key, ttl);
+            }
+
             let event = EoaExecutorEvent {
                 transaction_id: pending_transaction.transaction_id.clone(),
                 address: pending_transaction.user_request.from,
@@ -613,12 +627,14 @@ impl AtomicEoaExecutorStore {
         &self,
         results: Vec<SubmissionResult>,
         webhook_queue: Arc<twmq::Queue<WebhookJobHandler>>,
+        failed_transaction_expiry_seconds: u64,
     ) -> Result<BorrowedProcessingReport, TransactionStoreError> {
         self.execute_with_watch_and_retry(&ProcessBorrowedTransactions {
             results,
             keys: &self.keys,
             webhook_queue,
             eoa_metrics: &self.eoa_metrics,
+            failed_transaction_expiry_seconds,
         })
         .await
     }

executors/src/eoa/store/atomic.rs

@@ -12,7 +12,7 @@ use crate::eoa::{
     },
     worker::error::EoaExecutorWorkerError,
 };
-use crate::metrics::{current_timestamp_ms, calculate_duration_seconds, EoaMetrics};
+use crate::metrics::{EoaMetrics, calculate_duration_seconds, current_timestamp_ms};
 use crate::webhook::{WebhookJobHandler, queue_webhook_envelopes};
 
 #[derive(Debug, Clone)]
@@ -41,6 +41,7 @@ pub struct ProcessBorrowedTransactions<'a> {
     pub keys: &'a EoaExecutorStoreKeys,
     pub webhook_queue: Arc<Queue<WebhookJobHandler>>,
     pub eoa_metrics: &'a EoaMetrics,
+    pub failed_transaction_expiry_seconds: u64,
 }
 
 #[derive(Debug, Default)]
@@ -121,15 +122,13 @@ impl SafeRedisTransaction for ProcessBorrowedTransactions<'_> {
                 SubmissionResultType::Success => {
                     // Record metrics: transaction queued to sent
                     let sent_timestamp = current_timestamp_ms();
-                    let queued_to_sent_duration = calculate_duration_seconds(
-                        result.transaction.queued_at, 
-                        sent_timestamp
-                    );
+                    let queued_to_sent_duration =
+                        calculate_duration_seconds(result.transaction.queued_at, sent_timestamp);
                     // Record metrics using the clean EoaMetrics abstraction
                     self.eoa_metrics.record_transaction_sent(
                         self.keys.eoa,
                         self.keys.chain_id,
-                        queued_to_sent_duration
+                        queued_to_sent_duration,
                     );
 
                     // Add to submitted zset
@@ -189,7 +188,7 @@ impl SafeRedisTransaction for ProcessBorrowedTransactions<'_> {
                     // Update transaction data status
                     let tx_data_key = self.keys.transaction_data_key_name(transaction_id);
                     pipeline.hset(&tx_data_key, "status", "pending");
-                    
+
                     // ask for this nonce to be recycled because we did not consume the nonce
                     pipeline.zadd(self.keys.recycled_nonces_zset_name(), nonce, nonce);
 
@@ -219,12 +218,22 @@ impl SafeRedisTransaction for ProcessBorrowedTransactions<'_> {
                     report.moved_to_pending += 1;
                 }
                 SubmissionResultType::Fail(err) => {
-                    // Mark as failed
+                    // Mark as failed and set expiry
                     let tx_data_key = self.keys.transaction_data_key_name(transaction_id);
                     pipeline.hset(&tx_data_key, "status", "failed");
                     pipeline.hset(&tx_data_key, "completed_at", now);
                     pipeline.hset(&tx_data_key, "failure_reason", err.to_string());
 
+                    let ttl: i64 = i64::try_from(self.failed_transaction_expiry_seconds)
+                        .unwrap_or(i64::MAX)
+                        .max(1);
+
+                    // Set expiry on all related keys for failed transactions
+                    let transaction_keys = self.keys.get_all_transaction_keys(transaction_id);
+                    for key in transaction_keys {
+                        pipeline.expire(&key, ttl);
+                    }
+
                     // ask for this nonce to be recycled because we did not consume the nonce
                     pipeline.zadd(self.keys.recycled_nonces_zset_name(), nonce, nonce);
 

executors/src/eoa/store/borrowed.rs

@@ -186,6 +186,26 @@ impl EoaExecutorStoreKeys {
         }
     }
 
+    /// Get all Redis keys related to a transaction for cleanup
+    pub fn get_all_transaction_keys(&self, transaction_id: &str) -> Vec<String> {
+        vec![
+            self.transaction_data_key_name(transaction_id),
+            self.transaction_attempts_list_name(transaction_id),
+        ]
+    }
+
+    /// Get all Redis keys related to a transaction including hash mappings for cleanup
+    pub fn get_all_transaction_keys_with_hashes(&self, transaction_id: &str, transaction_hashes: &[String]) -> Vec<String> {
+        let mut keys = self.get_all_transaction_keys(transaction_id);
+        
+        // Add hash-to-id mappings
+        for hash in transaction_hashes {
+            keys.push(self.transaction_hash_to_id_key_name(hash));
+        }
+        
+        keys
+    }
+
     /// Name of the hashmap that maps `transaction_id` -> `BorrowedTransactionData`
     ///
     /// This is used for crash recovery. Before submitting a transaction, we atomically move from pending to this borrowed hashmap.

executors/src/eoa/store/mod.rs

@@ -354,21 +354,19 @@ impl SafeRedisTransaction for CleanSubmittedTransactions<'_> {
                     (id, Some(confirmed_tx)) => {
                         // Clean up confirmed transaction from Redis
                         self.remove_transaction_from_redis_submitted_zset(pipeline, tx);
-                        let data_key_name = self.keys.transaction_data_key_name(id);
-                        pipeline.hset(&data_key_name, "status", "confirmed");
-                        pipeline.hset(&data_key_name, "completed_at", now);
-                        pipeline.hset(
-                            &data_key_name,
-                            "receipt",
-                            confirmed_tx.receipt_serialized.clone(),
-                        );
+
+                        // IMMEDIATE CLEANUP: Delete all transaction data since it's confirmed
+                        // Note: Hash mappings will be cleaned up periodically by maintenance script
+                        let keys_to_delete = self.keys.get_all_transaction_keys(id);
+                        for key in keys_to_delete {
+                            pipeline.del(&key);
+                        }
 
                         if let SubmittedTransactionHydrated::Real(tx) = tx {
                             // Record metrics: transaction queued to mined for confirmed transactions
                             let confirmed_timestamp = current_timestamp_ms();
                             let queued_to_mined_duration =
                                 calculate_duration_seconds(tx.queued_at, confirmed_timestamp);
-                            // Record metrics using the clean EoaMetrics abstraction
                             self.eoa_metrics.record_transaction_confirmed(
                                 self.keys.eoa,
                                 self.keys.chain_id,

executors/src/eoa/store/submitted.rs

@@ -1,6 +1,7 @@
 use alloy::{primitives::B256, providers::Provider};
 use engine_core::{chain::Chain, error::AlloyRpcErrorToEngineError};
 use serde::{Deserialize, Serialize};
+use tracing::Instrument;
 
 use crate::{
     FlashblocksTransactionCount, TransactionCounts,
@@ -108,8 +109,38 @@ impl<C: Chain> EoaExecutorWorker<C> {
                     {
                         tracing::warn!(
                             error = ?e,
-                            "Failed to attempt gas bump for stalled nonce"
+                            "Failed to attempt gas bump for stalled nonce, trying fallback"
                         );
+                        
+                        // Fallback: try to send a no-op transaction
+                        tracing::info!(
+                            nonce = transaction_counts.preconfirmed,
+                            "Gas bump failed, attempting no-op transaction as fallback"
+                        );
+                        if let Ok(noop_tx) = self.send_noop_transaction(transaction_counts.preconfirmed).await {
+                            if let Err(e) = self.store.process_noop_transactions(&[noop_tx]).await {
+                                tracing::error!(
+                                    error = ?e,
+                                    "Failed to process fallback no-op transaction for stalled nonce"
+                                );
+                            }
+                        } else {
+                            tracing::error!("Failed to send fallback no-op transaction for stalled nonce");
+                            
+                            // Ultimate fallback: check if we should trigger auto-reset
+                            let time_since_movement = now.saturating_sub(current_health.last_nonce_movement_at);
+                            if time_since_movement > 5 * 60 * 1000 && submitted_count > 0 { // 5 minutes
+                                tracing::warn!(
+                                    nonce = transaction_counts.preconfirmed,
+                                    time_since_movement = time_since_movement,
+                                    "EOA appears permanently stuck, scheduling auto-reset"
+                                );
+                                
+                                if let Err(e) = self.store.schedule_manual_reset().await {
+                                    tracing::error!(error = ?e, "Failed to schedule auto-reset");
+                                }
+                            }
+                        }
                     }
                 }
             }
@@ -356,7 +387,7 @@ impl<C: Chain> EoaExecutorWorker<C> {
                         error = ?e,
                         "Failed to build typed transaction for gas bump"
                     );
-                    return Ok(false);
+                    return Err(e);
                 }
             };
             let bumped_typed_tx = self.apply_gas_bump_to_typed_transaction(typed_tx, 120); // 20% increase
@@ -375,7 +406,7 @@ impl<C: Chain> EoaExecutorWorker<C> {
                         error = ?e,
                         "Failed to sign transaction for gas bump"
                     );
-                    return Ok(false);
+                    return Err(e);
                 }
             };
 
@@ -393,9 +424,15 @@ impl<C: Chain> EoaExecutorWorker<C> {
                 )
                 .await?;
 
-            // Send the bumped transaction
-            let tx_envelope = bumped_tx.into();
-            match self.chain.provider().send_tx_envelope(tx_envelope).await {
+            // Send the bumped transaction with retry logic
+            match self.send_tx_envelope_with_retry(bumped_tx.into(), crate::eoa::worker::error::SendContext::InitialBroadcast)
+                .instrument(tracing::info_span!(
+                    "send_tx_envelope_with_retry", 
+                    transaction_id = %newest_transaction_data.transaction_id,
+                    nonce = expected_nonce,
+                    context = "gas_bump"
+                ))
+                .await {
                 Ok(_) => {
                     tracing::info!(
                         transaction_id = ?newest_transaction_data.transaction_id,
@@ -409,10 +446,13 @@ impl<C: Chain> EoaExecutorWorker<C> {
                         transaction_id = ?newest_transaction_data.transaction_id,
                         nonce = expected_nonce,
                         error = ?e,
-                        "Failed to send gas bumped transaction"
+                        "Failed to send gas bumped transaction after retries"
                     );
-                    // Don't fail the worker, just log the error
-                    Ok(false)
+                    // Convert RPC error to worker error and bubble up
+                    Err(EoaExecutorWorkerError::TransactionSendError {
+                        message: format!("Failed to send gas bumped transaction: {e:?}"),
+                        inner_error: e.to_engine_error(&self.chain),
+                    })
                 }
             }
         } else {