Fail stale jobs and cull zombie pending txns (#106)

* Fail stale jobs and cull zombie pending txns

Bound job lifetimes and remove stale pending transactions to prevent zombie retries and unbounded resource growth. Adds a 24h max age check for confirmation and send jobs (EIP-7702 and external bundler), including a job_age_seconds helper and logging, so long-lived retrying jobs are permanently failed. Implements peek_pending_transactions_older_than in the EOA store to fetch pending entries older than a cutoff (and clean up missing data). Adds EOA worker logic to cull stale pending transactions (24h cutoff, max 500 per cycle) by batch-failing them and enqueuing failure webhooks. Small logging/error messages added to surface these events.

* Add StaleJob error and use inspect_err

Replace manual if-let logging with combinators in EoaExecutorWorker: call cull_stale_pending_transactions().await.inspect_err(...).map_err(|e| e.handle()) to log errors and convert them via handle().

Introduce a new UserOpConfirmationError::StaleJob variant carrying user_op_hash, attempt_number, and age_seconds to represent confirmation jobs that aged out; return this variant when a job exceeds MAX_CONFIRMATION_JOB_AGE_SECONDS and update the error-to-message mapping to include a message for StaleJob.

* Extract hydrate_pending_transactions helper

Introduce hydrate_pending_transactions to centralize the logic for pipelined HGET of transaction user_request fields, JSON deserialization into PendingTransaction, and cleanup (ZREM) of orphaned zset entries. Replace duplicated hydration code in several peek_pending_transactions* methods with calls to the new helper, allocate the result Vec with capacity, and adjust pipeline query calls accordingly. This refactor reduces duplication and consolidates deserialization/error handling and cleanup in one place for easier maintenance.

Author: 0xFirekeeper

executors/src/eip7702_executor/confirm.rs

@@ -39,6 +39,17 @@ fn transaction_hash_retry_delay(attempts: u32) -> Duration {
     }
 }
 
+/// Maximum age (in seconds) of a confirmation job before it is permanently failed.
+const MAX_CONFIRMATION_JOB_AGE_SECONDS: u64 = 24 * 60 * 60;
+
+fn job_age_seconds<T: Clone>(job: &BorrowedJob<T>) -> u64 {
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0);
+    now.saturating_sub(job.job.created_at)
+}
+
 // --- Job Payload ---
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
@@ -186,6 +197,25 @@ where
         let job_data = &job.job.data;
         let transaction_hash_delay = transaction_hash_retry_delay(job.attempts());
 
+        let age_seconds = job_age_seconds(job);
+        if age_seconds > MAX_CONFIRMATION_JOB_AGE_SECONDS {
+            tracing::error!(
+                bundler_transaction_id = job_data.bundler_transaction_id,
+                transaction_id = job_data.transaction_id,
+                attempts = job.attempts(),
+                age_seconds,
+                max_age_seconds = MAX_CONFIRMATION_JOB_AGE_SECONDS,
+                "EIP-7702 confirmation job exceeded max age, failing permanently"
+            );
+            return Err(Eip7702ConfirmationError::TransactionHashError {
+                message: format!(
+                    "Job exceeded maximum retry age of {MAX_CONFIRMATION_JOB_AGE_SECONDS}s (current age: {age_seconds}s, attempts: {attempts}); failing to prevent zombie retries",
+                    attempts = job.attempts()
+                ),
+            })
+            .map_err_fail();
+        }
+
         // 1. Get Chain
         let chain = self
             .chain_service

executors/src/eip7702_executor/send.rs

@@ -182,6 +182,27 @@ where
     ) -> JobResult<Self::Output, Self::ErrorData> {
         let job_data = &job.job.data;
 
+        let now_secs = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_secs())
+            .unwrap_or(0);
+        let age_seconds = now_secs.saturating_sub(job.job.created_at);
+        if age_seconds > 24 * 60 * 60 {
+            tracing::error!(
+                transaction_id = job_data.transaction_id,
+                attempts = job.attempts(),
+                age_seconds,
+                "EIP-7702 send job exceeded max age, failing permanently"
+            );
+            return Err(Eip7702SendError::InternalError {
+                message: format!(
+                    "Send job exceeded maximum retry age of 24h (current age: {age_seconds}s, attempts: {attempts}); failing to prevent zombie retries",
+                    attempts = job.attempts()
+                ),
+            })
+            .map_err_fail();
+        }
+
         // 1. Get Chain
         let chain = self
             .chain_service

executors/src/eoa/store/mod.rs

@@ -549,10 +549,11 @@ impl EoaExecutorStore {
         self.peek_pending_transactions_paginated(0, limit).await
     }
 
-    /// Peek at pending transactions with pagination support
-    pub async fn peek_pending_transactions_paginated(
+    /// Peek at pending transactions whose `queued_at` (unix ms) is strictly older than
+    /// `older_than_unix_ms`.
+    pub async fn peek_pending_transactions_older_than(
         &self,
-        offset: u64,
+        older_than_unix_ms: u64,
         limit: u64,
     ) -> Result<Vec<PendingTransaction>, TransactionStoreError> {
         if limit == 0 {
@@ -562,27 +563,48 @@ impl EoaExecutorStore {
         let pending_key = self.pending_transactions_zset_name();
         let mut conn = self.redis.clone();
 
-        // Use ZRANGE to peek without removing, with offset support
-        let start = offset as isize;
-        let stop = (offset + limit - 1) as isize;
-
+        let max_exclusive = format!("({older_than_unix_ms}");
         let transaction_ids: Vec<PendingTransactionStringWithQueuedAt> =
-            conn.zrange_withscores(&pending_key, start, stop).await?;
+            twmq::redis::cmd("ZRANGEBYSCORE")
+                .arg(&pending_key)
+                .arg(0)
+                .arg(&max_exclusive)
+                .arg("WITHSCORES")
+                .arg("LIMIT")
+                .arg(0)
+                .arg(limit as isize)
+                .query_async(&mut conn)
+                .await?;
+
+        self.hydrate_pending_transactions(&mut conn, transaction_ids)
+            .await
+    }
 
+    /// Given a list of (transaction_id, queued_at) tuples, fetch each transaction's
+    /// `user_request` via a pipelined HGET, deserialize into `PendingTransaction`,
+    /// and ZREM any entries whose transaction data has gone missing.
+    ///
+    /// Shared by the various `peek_pending_transactions*` methods to centralize
+    /// hydration, deserialization error handling, and cleanup of orphaned zset
+    /// entries.
+    async fn hydrate_pending_transactions(
+        &self,
+        conn: &mut ConnectionManager,
+        transaction_ids: Vec<PendingTransactionStringWithQueuedAt>,
+    ) -> Result<Vec<PendingTransaction>, TransactionStoreError> {
         if transaction_ids.is_empty() {
             return Ok(Vec::new());
         }
 
         let mut pipe = twmq::redis::pipe();
-
         for (transaction_id, _) in &transaction_ids {
             let tx_data_key = self.transaction_data_key_name(transaction_id);
             pipe.hget(&tx_data_key, "user_request");
         }
+        let user_requests: Vec<Option<String>> = pipe.query_async(conn).await?;
 
-        let user_requests: Vec<Option<String>> = pipe.query_async(&mut conn).await?;
-
-        let mut pending_transactions: Vec<PendingTransaction> = Vec::new();
+        let mut pending_transactions: Vec<PendingTransaction> =
+            Vec::with_capacity(transaction_ids.len());
         let mut deletion_pipe = twmq::redis::pipe();
 
         for ((transaction_id, queued_at), user_request) in
@@ -608,29 +630,36 @@ impl EoaExecutorStore {
         }
 
         if !deletion_pipe.is_empty() {
-            deletion_pipe.query_async::<()>(&mut conn).await?;
+            deletion_pipe.query_async::<()>(conn).await?;
         }
 
-        // let user_requests: Vec<EoaTransactionRequest> = user_requests
-        //     .into_iter()
-        //     .map(|user_request_json| serde_json::from_str(&user_request_json))
-        //     .collect::<Result<Vec<EoaTransactionRequest>, serde_json::Error>>()?;
-
-        // let pending_transactions: Vec<PendingTransaction> = transaction_ids
-        //     .into_iter()
-        //     .zip(user_requests)
-        //     .map(
-        //         |((transaction_id, queued_at), user_request)| PendingTransaction {
-        //             transaction_id,
-        //             queued_at,
-        //             user_request,
-        //         },
-        //     )
-        //     .collect();
-
         Ok(pending_transactions)
     }
 
+    /// Peek at pending transactions with pagination support
+    pub async fn peek_pending_transactions_paginated(
+        &self,
+        offset: u64,
+        limit: u64,
+    ) -> Result<Vec<PendingTransaction>, TransactionStoreError> {
+        if limit == 0 {
+            return Ok(Vec::new());
+        }
+
+        let pending_key = self.pending_transactions_zset_name();
+        let mut conn = self.redis.clone();
+
+        // Use ZRANGE to peek without removing, with offset support
+        let start = offset as isize;
+        let stop = (offset + limit - 1) as isize;
+
+        let transaction_ids: Vec<PendingTransactionStringWithQueuedAt> =
+            conn.zrange_withscores(&pending_key, start, stop).await?;
+
+        self.hydrate_pending_transactions(&mut conn, transaction_ids)
+            .await
+    }
+
     /// Peek at pending transactions and get optimistic nonce in a single operation
     /// This is optimized for the send flow to reduce Redis round-trips
     pub async fn peek_pending_transactions_with_optimistic_nonce(
@@ -661,47 +690,9 @@ impl EoaExecutorStore {
 
         let optimistic = optimistic_nonce.ok_or_else(|| self.nonce_sync_required_error())?;
 
-        if transaction_ids.is_empty() {
-            return Ok((Vec::new(), optimistic));
-        }
-
-        // Second pipeline: Get transaction data
-        let mut pipe = twmq::redis::pipe();
-        for (transaction_id, _) in &transaction_ids {
-            let tx_data_key = self.transaction_data_key_name(transaction_id);
-            pipe.hget(&tx_data_key, "user_request");
-        }
-
-        let user_requests: Vec<Option<String>> = pipe.query_async(&mut conn).await?;
-
-        let mut pending_transactions: Vec<PendingTransaction> = Vec::new();
-        let mut deletion_pipe = twmq::redis::pipe();
-
-        for ((transaction_id, queued_at), user_request) in
-            transaction_ids.into_iter().zip(user_requests)
-        {
-            match user_request {
-                Some(user_request) => {
-                    let user_request_parsed = serde_json::from_str(&user_request)?;
-                    pending_transactions.push(PendingTransaction {
-                        transaction_id,
-                        queued_at,
-                        user_request: user_request_parsed,
-                    });
-                }
-                None => {
-                    tracing::warn!(
-                        "Transaction {} data was missing, deleting transaction from redis",
-                        transaction_id
-                    );
-                    deletion_pipe.zrem(self.keys.pending_transactions_zset_name(), transaction_id);
-                }
-            }
-        }
-
-        if !deletion_pipe.is_empty() {
-            deletion_pipe.query_async::<()>(&mut conn).await?;
-        }
+        let pending_transactions = self
+            .hydrate_pending_transactions(&mut conn, transaction_ids)
+            .await?;
 
         Ok((pending_transactions, optimistic))
     }

executors/src/eoa/worker/mod.rs

@@ -38,6 +38,9 @@ const MAX_RECYCLED_THRESHOLD: u64 = 50; // Circuit breaker from spec
 const TARGET_TRANSACTIONS_PER_EOA: u64 = 10; // Fleet management from spec
 const MIN_TRANSACTIONS_PER_EOA: u64 = 1; // Fleet management from spec
 
+const MAX_PENDING_TRANSACTION_AGE_MS: u64 = 24 * 60 * 60 * 1000;
+const MAX_STALE_PENDING_PER_CYCLE: u64 = 500;
+
 // ========== JOB DATA ==========
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
@@ -308,6 +311,18 @@ impl<C: Chain> EoaExecutorWorker<C> {
     async fn execute_main_workflow(
         &self,
     ) -> JobResult<EoaExecutorWorkerResult, EoaExecutorWorkerError> {
+        self.cull_stale_pending_transactions()
+            .await
+            .inspect_err(|e| {
+                tracing::error!(
+                    error = ?e,
+                    eoa = ?self.eoa,
+                    chain_id = self.chain_id,
+                    "Error culling stale pending transactions"
+                );
+            })
+            .map_err(|e| e.handle())?;
+
         // 1. CRASH RECOVERY
         let start_time = current_timestamp_ms();
         let recovered = self
@@ -402,6 +417,55 @@ impl<C: Chain> EoaExecutorWorker<C> {
         })
     }
 
+    // ========== STALE TRANSACTION CULLING ==========
+    #[tracing::instrument(skip_all)]
+    async fn cull_stale_pending_transactions(&self) -> Result<(), EoaExecutorWorkerError> {
+        let now_ms = current_timestamp_ms();
+        let cutoff = now_ms.saturating_sub(MAX_PENDING_TRANSACTION_AGE_MS);
+
+        let stale = self
+            .store
+            .peek_pending_transactions_older_than(cutoff, MAX_STALE_PENDING_PER_CYCLE)
+            .await
+            .map_err(EoaExecutorWorkerError::from)?;
+
+        if stale.is_empty() {
+            return Ok(());
+        }
+
+        tracing::warn!(
+            eoa = ?self.eoa,
+            chain_id = self.chain_id,
+            count = stale.len(),
+            cutoff_unix_ms = cutoff,
+            max_age_ms = MAX_PENDING_TRANSACTION_AGE_MS,
+            "Culling stale pending transactions older than max age"
+        );
+
+        let failures: Vec<(&crate::eoa::store::PendingTransaction, EoaExecutorWorkerError)> = stale
+            .iter()
+            .map(|p| {
+                let age_ms = now_ms.saturating_sub(p.queued_at);
+                (
+                    p,
+                    EoaExecutorWorkerError::InternalError {
+                        message: format!(
+                            "Transaction exceeded maximum pending age of {}ms (was pending for {}ms); failing to prevent zombie retries",
+                            MAX_PENDING_TRANSACTION_AGE_MS, age_ms
+                        ),
+                    },
+                )
+            })
+            .collect();
+
+        self.store
+            .fail_pending_transactions_batch(failures, self.webhook_queue.clone())
+            .await
+            .map_err(EoaExecutorWorkerError::from)?;
+
+        Ok(())
+    }
+
     // ========== CRASH RECOVERY ==========
     #[tracing::instrument(skip_all)]
     async fn recover_borrowed_state(&self) -> Result<u32, EoaExecutorWorkerError> {