Enhance metrics tracking and add dependencies (#34)

* Enhance metrics tracking and add dependencies

- Introduced `prometheus` and `lazy_static` dependencies to facilitate metrics collection.
- Added metrics recording for transaction durations in various components, including EIP-7702 and external bundler.
- Updated the `metrics` module to include functions for tracking transaction states and processing times.
- Enhanced the server to initialize and expose metrics via a dedicated endpoint.

These changes improve observability and performance monitoring across the execution framework.

* Implement TWMQ duration calculation for metrics tracking

- Added a new helper function `calculate_duration_seconds_from_twmq` to compute duration in seconds using timestamps in TWMQ format.
- Updated various components (EIP-7702 executor, external bundler, and EOA worker) to utilize the new TWMQ duration calculation for recording transaction metrics.
- Enhanced test coverage for the new duration calculation function.

These changes improve the accuracy of metrics tracking related to transaction durations.

* correct EOA worker time metrics

Author: d4mr

Cargo.lock

@@ -0,0 +1,38 @@
+version: "3.8"
+
+services:
+  prometheus:
+    image: prom/prometheus:latest
+    container_name: tw-engine-prometheus
+    ports:
+      - "9090:9090"
+    volumes:
+      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
+      - prometheus-data:/prometheus
+    command:
+      - "--config.file=/etc/prometheus/prometheus.yml"
+      - "--storage.tsdb.path=/prometheus"
+      - "--web.console.libraries=/etc/prometheus/console_libraries"
+      - "--web.console.templates=/etc/prometheus/consoles"
+      - "--storage.tsdb.retention.time=200h"
+      - "--web.enable-lifecycle"
+    restart: unless-stopped
+
+  grafana:
+    image: grafana/grafana:latest
+    container_name: tw-engine-grafana
+    ports:
+      - "3000:3000"
+    volumes:
+      - grafana-data:/var/lib/grafana
+      - ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards
+      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning
+    environment:
+      - GF_SECURITY_ADMIN_USER=admin
+      - GF_SECURITY_ADMIN_PASSWORD=admin
+      - GF_USERS_ALLOW_SIGN_UP=false
+    restart: unless-stopped
+
+volumes:
+  prometheus-data:
+  grafana-data:

docker-compose.monitoring.yml

@@ -25,3 +25,5 @@ chrono = "0.4.41"
 tokio = { version = "1.45.0", features = ["full"] }
 futures = "0.3.31"
 moka = { version = "0.12.10", features = ["future"] }
+prometheus = "0.13.4"
+lazy_static = "1.5.0"

executors/Cargo.toml

@@ -18,6 +18,7 @@ use twmq::{
 
 use crate::eip7702_executor::send::Eip7702Sender;
 use crate::{
+    metrics::{record_transaction_queued_to_confirmed, current_timestamp_ms, calculate_duration_seconds_from_twmq},
     transaction_registry::TransactionRegistry,
     webhook::{
         WebhookJobHandler,
@@ -37,6 +38,9 @@ pub struct Eip7702ConfirmationJobData {
     pub rpc_credentials: RpcCredentials,
     #[serde(default)]
     pub webhook_options: Vec<WebhookOptions>,
+    /// Original timestamp when the transaction was first queued (unix timestamp in milliseconds)
+    #[serde(default)]
+    pub original_queued_timestamp: Option<u64>,
 }
 
 impl HasWebhookOptions for Eip7702ConfirmationJobData {
@@ -262,6 +266,13 @@ where
             "Transaction confirmed successfully"
         );
 
+                    // Record metrics if original timestamp is available
+            if let Some(original_timestamp) = job_data.original_queued_timestamp {
+                let confirmed_timestamp = current_timestamp_ms();
+                let queued_to_confirmed_duration = calculate_duration_seconds_from_twmq(original_timestamp, confirmed_timestamp);
+                record_transaction_queued_to_confirmed("eip7702", job_data.chain_id, queued_to_confirmed_duration);
+            }
+
         Ok(Eip7702ConfirmationResult {
             transaction_id: job_data.transaction_id.clone(),
             transaction_hash,

executors/src/eip7702_executor/confirm.rs

@@ -22,6 +22,7 @@ use twmq::{
 };
 
 use crate::{
+    metrics::{record_transaction_queued_to_sent, current_timestamp_ms, calculate_duration_seconds_from_twmq},
     transaction_registry::TransactionRegistry,
     webhook::{
         WebhookJobHandler,
@@ -280,6 +281,11 @@ where
 
         tracing::debug!(transaction_id = ?transaction_id, "EIP-7702 transaction sent to bundler");
 
+                    // Record metrics: transaction queued to sent
+            let sent_timestamp = current_timestamp_ms();
+            let queued_to_sent_duration = calculate_duration_seconds_from_twmq(job.job.created_at, sent_timestamp);
+            record_transaction_queued_to_sent("eip7702", job_data.chain_id, queued_to_sent_duration);
+
         let sender_details = match session_key_target_address {
             Some(target_address) => Eip7702Sender::SessionKey {
                 session_key_address: owner_address,
@@ -327,6 +333,7 @@ where
                 sender_details: success_data.result.sender_details.clone(),
                 rpc_credentials: job.job.data.rpc_credentials.clone(),
                 webhook_options: job.job.data.webhook_options.clone(),
+                original_queued_timestamp: Some(job.job.created_at),
             })
             .with_id(job.job.transaction_id());
 

executors/src/eip7702_executor/send.rs

@@ -12,6 +12,7 @@ use crate::eoa::{
     },
     worker::error::EoaExecutorWorkerError,
 };
+use crate::metrics::{record_transaction_queued_to_sent, current_timestamp_ms, calculate_duration_seconds};
 use crate::webhook::{WebhookJobHandler, queue_webhook_envelopes};
 
 #[derive(Debug, Clone)]
@@ -117,6 +118,14 @@ impl SafeRedisTransaction for ProcessBorrowedTransactions<'_> {
 
             match &result.result {
                 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
+                    );
+                    record_transaction_queued_to_sent("eoa", self.keys.chain_id, queued_to_sent_duration);
+
                     // Add to submitted zset
                     let (submitted_tx_redis_string, nonce) =
                         result.transaction.to_redis_string_with_nonce();

executors/src/eoa/store/borrowed.rs

@@ -16,6 +16,7 @@ use crate::{
             TransactionStoreError, atomic::SafeRedisTransaction,
         },
     },
+    metrics::{record_transaction_queued_to_confirmed, current_timestamp_ms, calculate_duration_seconds},
     webhook::{WebhookJobHandler, queue_webhook_envelopes},
 };
 
@@ -329,6 +330,13 @@ impl SafeRedisTransaction for CleanSubmittedTransactions<'_> {
                         );
 
                         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_transaction_queued_to_confirmed("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/store/submitted.rs

@@ -1,7 +1,6 @@
 use alloy::{primitives::B256, providers::Provider};
 use engine_core::{chain::Chain, error::AlloyRpcErrorToEngineError};
 use serde::{Deserialize, Serialize};
-use twmq::redis::AsyncCommands;
 
 use crate::eoa::{
     EoaExecutorStore,

executors/src/eoa/worker/confirm.rs

@@ -23,6 +23,7 @@ use crate::eoa::authorization_cache::EoaAuthorizationCache;
 use crate::eoa::store::{
     AtomicEoaExecutorStore, EoaExecutorStore, EoaExecutorStoreKeys, EoaHealth, SubmissionResult,
 };
+use crate::metrics::{calculate_duration_seconds, calculate_duration_seconds_from_twmq, current_timestamp_ms, record_eoa_job_processing_time};
 use crate::webhook::WebhookJobHandler;
 
 pub mod confirm;
@@ -188,11 +189,17 @@ where
             signer: self.eoa_signer.clone(),
         };
 
+        let job_start_time = current_timestamp_ms();
         let result = worker.execute_main_workflow().await?;
         if let Err(e) = worker.release_eoa_lock().await {
             tracing::error!(error = ?e, "Error releasing EOA lock");
         }
 
+        // Record EOA job processing metrics
+        let job_end_time = current_timestamp_ms();
+        let job_duration = calculate_duration_seconds(job_start_time, job_end_time);
+        record_eoa_job_processing_time(data.chain_id, job_duration);
+
         if result.is_work_remaining() {
             Err(EoaExecutorWorkerError::WorkRemaining { result })
                 .map_err_nack(Some(Duration::from_secs(2)), RequeuePosition::Last)

executors/src/eoa/worker/mod.rs

@@ -15,6 +15,7 @@ use twmq::{
 };
 
 use crate::{
+    metrics::{record_transaction_queued_to_confirmed, current_timestamp_ms, calculate_duration_seconds_from_twmq},
     transaction_registry::TransactionRegistry,
     webhook::{
         WebhookJobHandler,
@@ -36,6 +37,9 @@ pub struct UserOpConfirmationJobData {
     pub deployment_lock_acquired: bool,
     pub webhook_options: Vec<WebhookOptions>,
     pub rpc_credentials: RpcCredentials,
+    /// Original timestamp when the transaction was first queued (unix timestamp in milliseconds)
+    #[serde(default)]
+    pub original_queued_timestamp: Option<u64>,
 }
 
 // --- Success Result ---
@@ -211,7 +215,14 @@ where
             "User operation confirmed on-chain"
         );
 
-        // 4. Success! Lock cleanup will happen atomically in on_success hook
+                    // 4. Record metrics if original timestamp is available
+            if let Some(original_timestamp) = job_data.original_queued_timestamp {
+                let confirmed_timestamp = current_timestamp_ms();
+                let queued_to_confirmed_duration = calculate_duration_seconds_from_twmq(original_timestamp, confirmed_timestamp);
+                record_transaction_queued_to_confirmed("erc4337-external", job_data.chain_id, queued_to_confirmed_duration);
+            }
+
+        // 5. Success! Lock cleanup will happen atomically in on_success hook
         Ok(UserOpConfirmationResult {
             user_op_hash: job_data.user_op_hash.clone(),
             receipt,