From cca2e05c4cc28c59a40c6735152fe9b5c50b6f52 Mon Sep 17 00:00:00 2001 From: 0xFirekeeper <0xFirekeeper@gmail.com> Date: Tue, 21 Apr 2026 17:47:38 +0700 Subject: [PATCH 1/3] 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. --- executors/src/eip7702_executor/confirm.rs | 30 ++++++++++ executors/src/eip7702_executor/send.rs | 21 +++++++ executors/src/eoa/store/mod.rs | 71 +++++++++++++++++++++++ executors/src/eoa/worker/mod.rs | 61 +++++++++++++++++++ executors/src/external_bundler/confirm.rs | 27 +++++++++ executors/src/external_bundler/send.rs | 21 +++++++ 6 files changed, 231 insertions(+) diff --git a/executors/src/eip7702_executor/confirm.rs b/executors/src/eip7702_executor/confirm.rs index ad15036..1da583f 100644 --- a/executors/src/eip7702_executor/confirm.rs +++ b/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(job: &BorrowedJob) -> 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 diff --git a/executors/src/eip7702_executor/send.rs b/executors/src/eip7702_executor/send.rs index f531eab..535b716 100644 --- a/executors/src/eip7702_executor/send.rs +++ b/executors/src/eip7702_executor/send.rs @@ -182,6 +182,27 @@ where ) -> JobResult { 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 diff --git a/executors/src/eoa/store/mod.rs b/executors/src/eoa/store/mod.rs index 2cb59b4..32d8a01 100644 --- a/executors/src/eoa/store/mod.rs +++ b/executors/src/eoa/store/mod.rs @@ -549,6 +549,77 @@ impl EoaExecutorStore { self.peek_pending_transactions_paginated(0, limit).await } + /// 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, + older_than_unix_ms: u64, + limit: u64, + ) -> Result, TransactionStoreError> { + if limit == 0 { + return Ok(Vec::new()); + } + + let pending_key = self.pending_transactions_zset_name(); + let mut conn = self.redis.clone(); + + let max_exclusive = format!("({older_than_unix_ms}"); + let transaction_ids: Vec = 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?; + + 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> = pipe.query_async(&mut conn).await?; + + let mut pending_transactions: Vec = 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?; + } + + Ok(pending_transactions) + } + /// Peek at pending transactions with pagination support pub async fn peek_pending_transactions_paginated( &self, diff --git a/executors/src/eoa/worker/mod.rs b/executors/src/eoa/worker/mod.rs index 33e47bf..fba21b5 100644 --- a/executors/src/eoa/worker/mod.rs +++ b/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,15 @@ impl EoaExecutorWorker { async fn execute_main_workflow( &self, ) -> JobResult { + if let Err(e) = self.cull_stale_pending_transactions().await { + tracing::error!( + error = ?e, + eoa = ?self.eoa, + chain_id = self.chain_id, + "Error culling stale pending transactions, continuing with main workflow" + ); + } + // 1. CRASH RECOVERY let start_time = current_timestamp_ms(); let recovered = self @@ -402,6 +414,55 @@ impl EoaExecutorWorker { }) } + // ========== 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 { diff --git a/executors/src/external_bundler/confirm.rs b/executors/src/external_bundler/confirm.rs index 2da468d..1f164ce 100644 --- a/executors/src/external_bundler/confirm.rs +++ b/executors/src/external_bundler/confirm.rs @@ -28,6 +28,16 @@ use crate::{ use super::deployment::RedisDeploymentLock; +const MAX_CONFIRMATION_JOB_AGE_SECONDS: u64 = 24 * 60 * 60; + +fn job_age_seconds(job: &BorrowedJob) -> 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")] @@ -154,6 +164,23 @@ where ) -> JobResult { let job_data = &job.job.data; + let age_seconds = job_age_seconds(job); + if age_seconds > MAX_CONFIRMATION_JOB_AGE_SECONDS { + tracing::error!( + transaction_id = job_data.transaction_id, + user_op_hash = ?job_data.user_op_hash, + attempts = job.job.attempts, + age_seconds, + max_age_seconds = MAX_CONFIRMATION_JOB_AGE_SECONDS, + "User-op confirmation job exceeded max age, failing permanently" + ); + return Err(UserOpConfirmationError::ReceiptNotAvailable { + user_op_hash: job_data.user_op_hash.clone(), + attempt_number: job.job.attempts, + }) + .map_err_fail(); + } + // 1. Get Chain let chain = self .chain_service diff --git a/executors/src/external_bundler/send.rs b/executors/src/external_bundler/send.rs index e869451..8a7b8e8 100644 --- a/executors/src/external_bundler/send.rs +++ b/executors/src/external_bundler/send.rs @@ -258,6 +258,27 @@ where ) -> JobResult { 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, + "External bundler send job exceeded max age, failing permanently" + ); + return Err(ExternalBundlerSendError::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 From 1cf1f240eb4101c91d44201fe01b36701acc2529 Mon Sep 17 00:00:00 2001 From: 0xFirekeeper <0xFirekeeper@gmail.com> Date: Tue, 21 Apr 2026 17:57:22 +0700 Subject: [PATCH 2/3] 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. --- executors/src/eoa/store/mod.rs | 23 +++++++++++------------ executors/src/eoa/worker/mod.rs | 19 +++++++++++-------- executors/src/external_bundler/confirm.rs | 16 +++++++++++++++- 3 files changed, 37 insertions(+), 21 deletions(-) diff --git a/executors/src/eoa/store/mod.rs b/executors/src/eoa/store/mod.rs index 32d8a01..27169bc 100644 --- a/executors/src/eoa/store/mod.rs +++ b/executors/src/eoa/store/mod.rs @@ -564,18 +564,17 @@ impl EoaExecutorStore { let mut conn = self.redis.clone(); let max_exclusive = format!("({older_than_unix_ms}"); - let transaction_ids: Vec = 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?; + let transaction_ids: Vec = + 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?; if transaction_ids.is_empty() { return Ok(Vec::new()); diff --git a/executors/src/eoa/worker/mod.rs b/executors/src/eoa/worker/mod.rs index fba21b5..fa7a344 100644 --- a/executors/src/eoa/worker/mod.rs +++ b/executors/src/eoa/worker/mod.rs @@ -311,14 +311,17 @@ impl EoaExecutorWorker { async fn execute_main_workflow( &self, ) -> JobResult { - if let Err(e) = self.cull_stale_pending_transactions().await { - tracing::error!( - error = ?e, - eoa = ?self.eoa, - chain_id = self.chain_id, - "Error culling stale pending transactions, continuing with main workflow" - ); - } + 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(); diff --git a/executors/src/external_bundler/confirm.rs b/executors/src/external_bundler/confirm.rs index 1f164ce..44dcbec 100644 --- a/executors/src/external_bundler/confirm.rs +++ b/executors/src/external_bundler/confirm.rs @@ -79,6 +79,16 @@ pub enum UserOpConfirmationError { attempt_number: u32, }, + #[error( + "Confirmation job aged out after {age_seconds}s (attempt {attempt_number}) for user operation {user_op_hash}" + )] + #[serde(rename_all = "camelCase")] + StaleJob { + user_op_hash: Bytes, + attempt_number: u32, + age_seconds: u64, + }, + #[error("Failed to query user operation receipt: {message}")] #[serde(rename_all = "camelCase")] ReceiptQueryFailed { @@ -174,9 +184,10 @@ where max_age_seconds = MAX_CONFIRMATION_JOB_AGE_SECONDS, "User-op confirmation job exceeded max age, failing permanently" ); - return Err(UserOpConfirmationError::ReceiptNotAvailable { + return Err(UserOpConfirmationError::StaleJob { user_op_hash: job_data.user_op_hash.clone(), attempt_number: job.job.attempts, + age_seconds, }) .map_err_fail(); } @@ -361,6 +372,9 @@ where UserOpConfirmationError::ReceiptNotAvailable { .. } => { "Max confirmation attempts exceeded" } + UserOpConfirmationError::StaleJob { .. } => { + "Confirmation job aged out after exceeding max retry lifetime" + } _ => "Confirmation job failed permanently", }; From f773afef6e4363fc9d10cdf8b5681f0a2636eed4 Mon Sep 17 00:00:00 2001 From: 0xFirekeeper <0xFirekeeper@gmail.com> Date: Tue, 21 Apr 2026 18:08:32 +0700 Subject: [PATCH 3/3] 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. --- executors/src/eoa/store/mod.rs | 129 +++++++-------------------------- 1 file changed, 25 insertions(+), 104 deletions(-) diff --git a/executors/src/eoa/store/mod.rs b/executors/src/eoa/store/mod.rs index 27169bc..6df694a 100644 --- a/executors/src/eoa/store/mod.rs +++ b/executors/src/eoa/store/mod.rs @@ -576,6 +576,22 @@ impl EoaExecutorStore { .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, + ) -> Result, TransactionStoreError> { if transaction_ids.is_empty() { return Ok(Vec::new()); } @@ -585,9 +601,10 @@ impl EoaExecutorStore { let tx_data_key = self.transaction_data_key_name(transaction_id); pipe.hget(&tx_data_key, "user_request"); } - let user_requests: Vec> = pipe.query_async(&mut conn).await?; + let user_requests: Vec> = pipe.query_async(conn).await?; - let mut pending_transactions: Vec = Vec::new(); + let mut pending_transactions: Vec = + Vec::with_capacity(transaction_ids.len()); let mut deletion_pipe = twmq::redis::pipe(); for ((transaction_id, queued_at), user_request) in @@ -613,7 +630,7 @@ impl EoaExecutorStore { } if !deletion_pipe.is_empty() { - deletion_pipe.query_async::<()>(&mut conn).await?; + deletion_pipe.query_async::<()>(conn).await?; } Ok(pending_transactions) @@ -639,66 +656,8 @@ impl EoaExecutorStore { let transaction_ids: Vec = conn.zrange_withscores(&pending_key, start, stop).await?; - 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> = pipe.query_async(&mut conn).await?; - - let mut pending_transactions: Vec = 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 user_requests: Vec = user_requests - // .into_iter() - // .map(|user_request_json| serde_json::from_str(&user_request_json)) - // .collect::, serde_json::Error>>()?; - - // let pending_transactions: Vec = 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) + self.hydrate_pending_transactions(&mut conn, transaction_ids) + .await } /// Peek at pending transactions and get optimistic nonce in a single operation @@ -731,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> = pipe.query_async(&mut conn).await?; - - let mut pending_transactions: Vec = 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)) }