Add EOA metrics tracking and abstraction for degradation and stuck conditions (#40)

- Introduced new `EoaMetrics` struct to encapsulate configuration and provide a clean interface for tracking EOA transaction metrics.
- Added metrics for EOA send degradation, confirmation degradation, and stuck conditions, including thresholds for each.
- Updated various components to utilize the new `EoaMetrics` abstraction for recording transaction metrics, enhancing observability and performance monitoring.
- Configured default thresholds in the server configuration for EOA metrics tracking.

These changes improve the robustness of EOA transaction monitoring and facilitate better handling of problematic EOAs.

Author: d4mr

executors/src/eoa/store/atomic.rs

@@ -68,6 +68,7 @@ pub trait SafeRedisTransaction: Send + Sync {
 pub struct AtomicEoaExecutorStore {
     pub store: EoaExecutorStore,
     pub worker_id: String,
+    pub eoa_metrics: crate::metrics::EoaMetrics,
 }
 
 impl std::ops::Deref for AtomicEoaExecutorStore {
@@ -600,6 +601,7 @@ impl AtomicEoaExecutorStore {
             last_confirmed_nonce,
             keys: &self.keys,
             webhook_queue,
+            eoa_metrics: &self.eoa_metrics,
         })
         .await
     }
@@ -616,6 +618,7 @@ impl AtomicEoaExecutorStore {
             results,
             keys: &self.keys,
             webhook_queue,
+            eoa_metrics: &self.eoa_metrics,
         })
         .await
     }

executors/src/eoa/store/borrowed.rs

@@ -12,7 +12,7 @@ use crate::eoa::{
     },
     worker::error::EoaExecutorWorkerError,
 };
-use crate::metrics::{record_transaction_queued_to_sent, current_timestamp_ms, calculate_duration_seconds};
+use crate::metrics::{current_timestamp_ms, calculate_duration_seconds, EoaMetrics};
 use crate::webhook::{WebhookJobHandler, queue_webhook_envelopes};
 
 #[derive(Debug, Clone)]
@@ -40,6 +40,7 @@ pub struct ProcessBorrowedTransactions<'a> {
     pub results: Vec<SubmissionResult>,
     pub keys: &'a EoaExecutorStoreKeys,
     pub webhook_queue: Arc<Queue<WebhookJobHandler>>,
+    pub eoa_metrics: &'a EoaMetrics,
 }
 
 #[derive(Debug, Default)]
@@ -124,7 +125,12 @@ impl SafeRedisTransaction for ProcessBorrowedTransactions<'_> {
                         result.transaction.queued_at, 
                         sent_timestamp
                     );
-                    record_transaction_queued_to_sent("eoa", self.keys.chain_id, queued_to_sent_duration);
+                    // Record metrics using the clean EoaMetrics abstraction
+                    self.eoa_metrics.record_transaction_sent(
+                        self.keys.eoa,
+                        self.keys.chain_id,
+                        queued_to_sent_duration
+                    );
 
                     // Add to submitted zset
                     let (submitted_tx_redis_string, nonce) =

executors/src/eoa/store/mod.rs

@@ -407,6 +407,7 @@ impl EoaExecutorStore {
     pub async fn acquire_eoa_lock_aggressively(
         self,
         worker_id: &str,
+        eoa_metrics: crate::metrics::EoaMetrics,
     ) -> Result<AtomicEoaExecutorStore, TransactionStoreError> {
         let lock_key = self.eoa_lock_key_name();
         let mut conn = self.redis.clone();
@@ -417,6 +418,7 @@ impl EoaExecutorStore {
             return Ok(AtomicEoaExecutorStore {
                 store: self,
                 worker_id: worker_id.to_string(),
+                eoa_metrics,
             });
         }
         let conflict_worker_id = conn.get::<_, Option<String>>(&lock_key).await?;
@@ -434,6 +436,7 @@ impl EoaExecutorStore {
         Ok(AtomicEoaExecutorStore {
             store: self,
             worker_id: worker_id.to_string(),
+            eoa_metrics,
         })
     }
 

executors/src/eoa/store/submitted.rs

@@ -16,7 +16,7 @@ use crate::{
             TransactionStoreError, atomic::SafeRedisTransaction,
         },
     },
-    metrics::{record_transaction_queued_to_confirmed, current_timestamp_ms, calculate_duration_seconds},
+    metrics::{current_timestamp_ms, calculate_duration_seconds, EoaMetrics},
     webhook::{WebhookJobHandler, queue_webhook_envelopes},
 };
 
@@ -198,6 +198,7 @@ pub struct CleanSubmittedTransactions<'a> {
     pub confirmed_transactions: &'a [ConfirmedTransaction],
     pub keys: &'a EoaExecutorStoreKeys,
     pub webhook_queue: Arc<twmq::Queue<WebhookJobHandler>>,
+    pub eoa_metrics: &'a EoaMetrics,
 }
 
 pub struct CleanAndGetRecycledNonces<'a> {
@@ -336,7 +337,12 @@ impl SafeRedisTransaction for CleanSubmittedTransactions<'_> {
                                 tx.queued_at, 
                                 confirmed_timestamp
                             );
-                            record_transaction_queued_to_confirmed("eoa", self.keys.chain_id, queued_to_mined_duration);
+                            // Record metrics using the clean EoaMetrics abstraction
+                            self.eoa_metrics.record_transaction_confirmed(
+                                self.keys.eoa,
+                                self.keys.chain_id,
+                                queued_to_mined_duration
+                            );
                             if !tx.user_request.webhook_options.is_empty() {
                                 let event = EoaExecutorEvent {
                                     transaction_id: tx.transaction_id.clone(),

executors/src/eoa/worker/confirm.rs

@@ -97,6 +97,25 @@ impl<C: Chain> EoaExecutorWorker<C> {
                     );
                 }
             }
+            
+            // Check if EOA is stuck and record metric using the clean EoaMetrics abstraction
+            let time_since_movement_seconds = time_since_movement as f64 / 1000.0;
+            if self.store.eoa_metrics.is_stuck(time_since_movement) {
+                tracing::warn!(
+                    time_since_movement = time_since_movement,
+                    stuck_threshold = self.store.eoa_metrics.stuck_threshold_seconds,
+                    eoa = ?self.eoa,
+                    chain_id = self.chain_id,
+                    "EOA is stuck - nonce hasn't moved for too long"
+                );
+                
+                // Record stuck EOA metric (low cardinality - only problematic EOAs)
+                self.store.eoa_metrics.record_stuck_eoa(
+                    self.eoa,
+                    self.chain_id,
+                    time_since_movement_seconds
+                );
+            }
 
             tracing::debug!("No nonce progress, still going ahead with confirm flow");
             // return Ok(CleanupReport::default());

executors/src/eoa/worker/mod.rs

@@ -24,7 +24,7 @@ use crate::eoa::store::{
     AtomicEoaExecutorStore, EoaExecutorStore, EoaExecutorStoreKeys, EoaHealth, SubmissionResult,
 };
 use crate::metrics::{
-    calculate_duration_seconds, current_timestamp_ms, record_eoa_job_processing_time,
+    EoaMetrics, calculate_duration_seconds, current_timestamp_ms, record_eoa_job_processing_time,
 };
 use crate::webhook::WebhookJobHandler;
 
@@ -123,6 +123,9 @@ where
     pub eoa_signer: Arc<EoaSigner>,
     pub max_inflight: u64, // Note: Spec uses MAX_INFLIGHT_PER_EOA constant
     pub max_recycled_nonces: u64, // Note: Spec uses MAX_RECYCLED_THRESHOLD constant
+
+    // EOA metrics abstraction with encapsulated configuration
+    pub eoa_metrics: EoaMetrics,
 }
 
 impl<CS> DurableExecution for EoaExecutorJobHandler<CS>
@@ -159,7 +162,7 @@ where
             data.eoa_address,
             data.chain_id,
         )
-        .acquire_eoa_lock_aggressively(&worker_id)
+        .acquire_eoa_lock_aggressively(&worker_id, self.eoa_metrics.clone())
         .await
         .map_err(|e| Into::<EoaExecutorWorkerError>::into(e).handle())?;
 

executors/src/metrics.rs

@@ -7,6 +7,10 @@ pub struct ExecutorMetrics {
     pub transaction_queued_to_sent_duration: HistogramVec,
     pub transaction_queued_to_confirmed_duration: HistogramVec,
     pub eoa_job_processing_duration: HistogramVec,
+    // EOA degradation and stuck metrics (low cardinality - only problematic EOAs)
+    pub eoa_degraded_send_duration: HistogramVec,
+    pub eoa_degraded_confirmation_duration: HistogramVec,
+    pub eoa_stuck_duration: HistogramVec,
 }
 
 impl ExecutorMetrics {
@@ -39,10 +43,41 @@ impl ExecutorMetrics {
             registry
         )?;
 
+        // EOA degradation and stuck metrics (low cardinality - only problematic EOAs)
+        let eoa_degraded_send_duration = register_histogram_vec_with_registry!(
+            HistogramOpts::new(
+                "tw_engine_executor_eoa_degraded_send_duration_seconds",
+                "Duration of EOA transactions that exceeded the send degradation threshold"
+            ).buckets(vec![5.0, 10.0, 20.0, 30.0, 60.0, 120.0, 300.0, 600.0]),
+            &["eoa_address", "chain_id"],
+            registry
+        )?;
+        
+        let eoa_degraded_confirmation_duration = register_histogram_vec_with_registry!(
+            HistogramOpts::new(
+                "tw_engine_executor_eoa_degraded_confirmation_duration_seconds",
+                "Duration of EOA transactions that exceeded the confirmation degradation threshold"
+            ).buckets(vec![30.0, 45.0, 60.0, 120.0, 300.0, 600.0, 1200.0, 1800.0, 3600.0]),
+            &["eoa_address", "chain_id"],
+            registry
+        )?;
+        
+        let eoa_stuck_duration = register_histogram_vec_with_registry!(
+            HistogramOpts::new(
+                "tw_engine_executor_eoa_stuck_duration_seconds",
+                "Duration since last nonce movement for EOAs that are considered stuck"
+            ).buckets(vec![200.0, 300.0, 600.0, 1200.0, 1800.0, 3600.0, 7200.0, 14400.0]),
+            &["eoa_address", "chain_id"],
+            registry
+        )?;
+
         Ok(ExecutorMetrics {
             transaction_queued_to_sent_duration,
             transaction_queued_to_confirmed_duration,
             eoa_job_processing_duration,
+            eoa_degraded_send_duration,
+            eoa_degraded_confirmation_duration,
+            eoa_stuck_duration,
         })
     }
 }
@@ -116,6 +151,76 @@ pub fn record_eoa_job_processing_time(chain_id: u64, duration_seconds: f64) {
         .observe(duration_seconds);
 }
 
+/// EOA Metrics abstraction that encapsulates configuration and provides clean interface
+#[derive(Debug, Clone)]
+pub struct EoaMetrics {
+    pub send_degradation_threshold_seconds: u64,
+    pub confirmation_degradation_threshold_seconds: u64,
+    pub stuck_threshold_seconds: u64,
+}
+
+impl EoaMetrics {
+    /// Create new EoaMetrics with configuration
+    pub fn new(
+        send_degradation_threshold_seconds: u64,
+        confirmation_degradation_threshold_seconds: u64,
+        stuck_threshold_seconds: u64,
+    ) -> Self {
+        Self {
+            send_degradation_threshold_seconds,
+            confirmation_degradation_threshold_seconds,
+            stuck_threshold_seconds,
+        }
+    }
+
+    /// Record EOA transaction send metrics with automatic degradation detection
+    pub fn record_transaction_sent(&self, eoa_address: alloy::primitives::Address, chain_id: u64, duration_seconds: f64) {
+        // Always record the regular metric
+        record_transaction_queued_to_sent("eoa", chain_id, duration_seconds);
+        
+        // Only record degraded metric if threshold exceeded (low cardinality)
+        if duration_seconds > self.send_degradation_threshold_seconds as f64 {
+            let metrics = get_metrics();
+            metrics.eoa_degraded_send_duration
+                .with_label_values(&[&eoa_address.to_string(), &chain_id.to_string()])
+                .observe(duration_seconds);
+        }
+    }
+
+    /// Record EOA transaction confirmation metrics with automatic degradation detection
+    pub fn record_transaction_confirmed(&self, eoa_address: alloy::primitives::Address, chain_id: u64, duration_seconds: f64) {
+        // Always record the regular metric
+        record_transaction_queued_to_confirmed("eoa", chain_id, duration_seconds);
+        
+        // Only record degraded metric if threshold exceeded (low cardinality)
+        if duration_seconds > self.confirmation_degradation_threshold_seconds as f64 {
+            let metrics = get_metrics();
+            metrics.eoa_degraded_confirmation_duration
+                .with_label_values(&[&eoa_address.to_string(), &chain_id.to_string()])
+                .observe(duration_seconds);
+        }
+    }
+
+    /// Record stuck EOA metric when nonce hasn't moved for too long
+    pub fn record_stuck_eoa(&self, eoa_address: alloy::primitives::Address, chain_id: u64, time_since_last_movement_seconds: f64) {
+        // Only record if EOA is actually stuck (exceeds threshold)
+        if time_since_last_movement_seconds > self.stuck_threshold_seconds as f64 {
+            let metrics = get_metrics();
+            metrics.eoa_stuck_duration
+                .with_label_values(&[&eoa_address.to_string(), &chain_id.to_string()])
+                .observe(time_since_last_movement_seconds);
+        }
+    }
+
+    /// Check if an EOA should be considered stuck based on time since last nonce movement
+    pub fn is_stuck(&self, time_since_last_movement_ms: u64) -> bool {
+        let time_since_last_movement_seconds = time_since_last_movement_ms as f64 / 1000.0;
+        time_since_last_movement_seconds > self.stuck_threshold_seconds as f64
+    }
+}
+
+
+
 /// Helper to calculate duration in seconds from unix timestamps (milliseconds)
 pub fn calculate_duration_seconds(start_timestamp_ms: u64, end_timestamp_ms: u64) -> f64 {
     (end_timestamp_ms.saturating_sub(start_timestamp_ms)) as f64 / 1000.0
@@ -165,11 +270,24 @@ mod tests {
         record_transaction_queued_to_confirmed("test", 1, 10.0);
         record_eoa_job_processing_time(1, 2.0);
 
+        // Test new EOA metrics abstraction
+        let eoa_metrics = EoaMetrics::new(10, 120, 600);
+        let test_address = "0x1234567890123456789012345678901234567890".parse().unwrap();
+        
+        eoa_metrics.record_transaction_sent(test_address, 1, 5.0); // Won't record degradation (below threshold)
+        eoa_metrics.record_transaction_sent(test_address, 1, 15.0); // Will record degradation (above threshold)
+        eoa_metrics.record_transaction_confirmed(test_address, 1, 60.0); // Won't record degradation (below threshold)
+        eoa_metrics.record_transaction_confirmed(test_address, 1, 180.0); // Will record degradation (above threshold)
+        eoa_metrics.record_stuck_eoa(test_address, 1, 900.0); // Will record stuck EOA
+        
         // Test that default metrics can be exported
         let metrics_output = export_default_metrics().expect("Should be able to export default metrics");
         assert!(metrics_output.contains("tw_engine_executor_transaction_queued_to_sent_duration_seconds"));
         assert!(metrics_output.contains("tw_engine_executor_transaction_queued_to_confirmed_duration_seconds"));
         assert!(metrics_output.contains("tw_engine_eoa_executor_job_processing_duration_seconds"));
+        assert!(metrics_output.contains("tw_engine_executor_eoa_degraded_send_duration_seconds"));
+        assert!(metrics_output.contains("tw_engine_executor_eoa_degraded_confirmation_duration_seconds"));
+        assert!(metrics_output.contains("tw_engine_executor_eoa_stuck_duration_seconds"));
     }
 
     #[test]

server/configuration/server_base.yaml

@@ -27,3 +27,7 @@ queue:
   always_poll: false
   max_success: 1000
   max_failed: 10000
+  monitoring:
+    eoa_send_degradation_threshold_seconds: 30
+    eoa_confirmation_degradation_threshold_seconds: 60
+    eoa_stuck_threshold_seconds: 300

server/src/config.rs

@@ -24,6 +24,27 @@ pub struct QueueConfig {
     pub local_concurrency: usize,
     pub polling_interval_ms: u64,
     pub lease_duration_seconds: u64,
+    
+    #[serde(default)]
+    pub monitoring: MonitoringConfig,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+#[serde(default)]
+pub struct MonitoringConfig {
+    pub eoa_send_degradation_threshold_seconds: u64,
+    pub eoa_confirmation_degradation_threshold_seconds: u64,
+    pub eoa_stuck_threshold_seconds: u64,
+}
+
+impl Default for MonitoringConfig {
+    fn default() -> Self {
+        Self {
+            eoa_send_degradation_threshold_seconds: 10, // 10 seconds
+            eoa_confirmation_degradation_threshold_seconds: 120, // 2 minutes
+            eoa_stuck_threshold_seconds: 600, // 10 minutes
+        }
+    }
 }
 
 #[derive(Debug, Clone, Deserialize)]

server/src/queue/manager.rs

@@ -207,6 +207,12 @@ impl QueueManager {
             .arc();
 
         // Create EOA executor queue
+        let eoa_metrics = engine_executors::metrics::EoaMetrics::new(
+            queue_config.monitoring.eoa_send_degradation_threshold_seconds,
+            queue_config.monitoring.eoa_confirmation_degradation_threshold_seconds,
+            queue_config.monitoring.eoa_stuck_threshold_seconds,
+        );
+        
         let eoa_executor_handler = EoaExecutorJobHandler {
             chain_service: chain_service.clone(),
             eoa_signer: eoa_signer.clone(),
@@ -216,6 +222,7 @@ impl QueueManager {
             authorization_cache,
             max_inflight: 100,
             max_recycled_nonces: 50,
+            eoa_metrics,
         };
 
         let eoa_executor_queue = Queue::builder()