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..6df694a 100644 --- a/executors/src/eoa/store/mod.rs +++ b/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, 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 = - 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, + ) -> Result, 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> = pipe.query_async(conn).await?; - let user_requests: Vec> = pipe.query_async(&mut 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 @@ -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 = 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) } + /// Peek at pending transactions with pagination support + pub async fn peek_pending_transactions_paginated( + &self, + offset: 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(); + + // 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 = + 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> = 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)) } diff --git a/executors/src/eoa/worker/mod.rs b/executors/src/eoa/worker/mod.rs index 33e47bf..fa7a344 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,18 @@ impl EoaExecutorWorker { async fn execute_main_workflow( &self, ) -> JobResult { + 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 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..44dcbec 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")] @@ -69,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 { @@ -154,6 +174,24 @@ 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::StaleJob { + user_op_hash: job_data.user_op_hash.clone(), + attempt_number: job.job.attempts, + age_seconds, + }) + .map_err_fail(); + } + // 1. Get Chain let chain = self .chain_service @@ -334,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", }; 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