Compute eoa worker ID independent of twmq lease token

executors/src/eoa/store/atomic.rs

@@ -11,8 +11,9 @@ use crate::{
         EoaExecutorStore,
         events::EoaExecutorEvent,
         store::{
-            BorrowedTransactionData, ConfirmedTransaction, EoaHealth, PendingTransaction,
-            SubmittedTransactionDehydrated, TransactionAttempt, TransactionStoreError,
+            BorrowedTransactionData, ConfirmedTransaction, EoaExecutorStoreKeys, EoaHealth,
+            PendingTransaction, SubmittedTransactionDehydrated, TransactionAttempt,
+            TransactionStoreError,
             borrowed::{BorrowedProcessingReport, ProcessBorrowedTransactions, SubmissionResult},
             pending::{
                 MovePendingToBorrowedWithIncrementedNonces, MovePendingToBorrowedWithRecycledNonces,
@@ -393,6 +394,7 @@ impl AtomicEoaExecutorStore {
     /// Synchronize nonces with the chain
     ///
     /// Part of standard nonce management flow, called in the confirm stage when chain nonce advances, and we need to update our cached nonce
+    #[tracing::instrument(skip_all, fields(current_chain_tx_count = current_chain_tx_count))]
     pub async fn update_cached_transaction_count(
         &self,
         current_chain_tx_count: u64,
@@ -502,47 +504,35 @@ impl AtomicEoaExecutorStore {
     ///
     /// This is called when we have too many recycled nonces and detect something wrong
     /// We want to start fresh, with the chain nonce as the new optimistic nonce
+    #[tracing::instrument(skip_all)]
     pub async fn reset_nonces(
         &self,
         current_chain_tx_count: u64,
     ) -> Result<(), TransactionStoreError> {
-        let now = chrono::Utc::now().timestamp_millis().max(0) as u64;
-
-        let current_health = self.get_eoa_health().await?;
-
-        // Prepare health update if health data exists
-        let health_update = if let Some(mut health) = current_health {
-            health.nonce_resets.push(now);
-            Some(serde_json::to_string(&health)?)
-        } else {
-            None
+        let reset_tx = ResetNoncesTransaction {
+            keys: &self.store.keys,
+            current_chain_tx_count,
         };
 
-        self.with_lock_check(|pipeline| {
-            let optimistic_key = self.optimistic_transaction_count_key_name();
-            let cached_nonce_key = self.last_transaction_count_key_name();
-            let recycled_key = self.recycled_nonces_zset_name();
-            let manual_reset_key = self.manual_reset_key_name();
+        let reset_result = self.execute_with_watch_and_retry(&reset_tx).await;
 
-            // Update health data only if it exists
-            if let Some(ref health_json) = health_update {
-                let health_key = self.eoa_health_key_name();
-                pipeline.set(&health_key, health_json);
+        match &reset_result {
+            Ok(()) => {
+                tracing::info!(
+                    current_chain_tx_count = current_chain_tx_count,
+                    "Reset nonces successfully"
+                );
             }
+            Err(e) => {
+                tracing::error!(
+                    current_chain_tx_count = current_chain_tx_count,
+                    error = ?e,
+                    "Failed to reset nonces"
+                );
+            }
+        }
 
-            // Reset the optimistic nonce
-            pipeline.set(&optimistic_key, current_chain_tx_count);
-
-            // Reset the cached nonce
-            pipeline.set(&cached_nonce_key, current_chain_tx_count);
-
-            // Reset the recycled nonces
-            pipeline.del(recycled_key);
-
-            // Delete the manual reset key
-            pipeline.del(&manual_reset_key);
-        })
-        .await
+        reset_result
     }
 
     /// Fail a transaction that's in the pending state (remove from pending and fail)
@@ -654,3 +644,79 @@ impl AtomicEoaExecutorStore {
         .await
     }
 }
+
+/// SafeRedisTransaction implementation for resetting nonces
+pub struct ResetNoncesTransaction<'a> {
+    pub keys: &'a EoaExecutorStoreKeys,
+    pub current_chain_tx_count: u64,
+}
+
+impl SafeRedisTransaction for ResetNoncesTransaction<'_> {
+    type ValidationData = Option<String>;
+    type OperationResult = ();
+
+    fn name(&self) -> &str {
+        "reset nonces"
+    }
+
+    fn watch_keys(&self) -> Vec<String> {
+        vec![
+            self.keys.optimistic_transaction_count_key_name(),
+            self.keys.last_transaction_count_key_name(),
+            self.keys.recycled_nonces_zset_name(),
+            self.keys.manual_reset_key_name(),
+        ]
+    }
+
+    async fn validation(
+        &self,
+        _conn: &mut ConnectionManager,
+        store: &EoaExecutorStore,
+    ) -> Result<Self::ValidationData, TransactionStoreError> {
+        let now = chrono::Utc::now().timestamp_millis().max(0) as u64;
+
+        // Get current health data to prepare update
+        let current_health = store.get_eoa_health().await?;
+        let health_update = if let Some(mut health) = current_health {
+            health.nonce_resets.push(now);
+            // Keep only the last 5 nonce reset timestamps
+            if health.nonce_resets.len() > 5 {
+                health.nonce_resets.drain(0..health.nonce_resets.len() - 5);
+            }
+            Some(serde_json::to_string(&health)?)
+        } else {
+            None
+        };
+
+        Ok(health_update)
+    }
+
+    fn operation(
+        &self,
+        pipeline: &mut twmq::redis::Pipeline,
+        health_update: Self::ValidationData,
+    ) -> Self::OperationResult {
+        let optimistic_key = self.keys.optimistic_transaction_count_key_name();
+        let cached_nonce_key = self.keys.last_transaction_count_key_name();
+        let recycled_key = self.keys.recycled_nonces_zset_name();
+        let manual_reset_key = self.keys.manual_reset_key_name();
+
+        // Update health data only if it exists
+        if let Some(ref health_json) = health_update {
+            let health_key = self.keys.eoa_health_key_name();
+            pipeline.set(&health_key, health_json);
+        }
+
+        // Reset the optimistic nonce
+        pipeline.set(&optimistic_key, self.current_chain_tx_count);
+
+        // Reset the cached nonce
+        pipeline.set(&cached_nonce_key, self.current_chain_tx_count);
+
+        // Reset the recycled nonces
+        pipeline.del(recycled_key);
+
+        // Delete the manual reset key
+        pipeline.del(&manual_reset_key);
+    }
+}

executors/src/eoa/store/mod.rs

@@ -284,7 +284,10 @@ impl EoaExecutorStoreKeys {
                 "{ns}:eoa_executor:pending_manual_reset:{}:{}",
                 self.chain_id, self.eoa
             ),
-            None => format!("eoa_executor:pending_manual_reset:{}:{}", self.chain_id, self.eoa),
+            None => format!(
+                "eoa_executor:pending_manual_reset:{}:{}",
+                self.chain_id, self.eoa
+            ),
         }
     }
 }
@@ -416,12 +419,15 @@ impl EoaExecutorStore {
                 worker_id: worker_id.to_string(),
             });
         }
+        let conflict_worker_id = conn.get::<_, String>(&lock_key).await?;
+
         // Lock exists, forcefully take it over
         tracing::warn!(
             eoa = ?self.eoa,
             chain_id = self.chain_id,
             worker_id = worker_id,
-            "Forcefully taking over EOA lock from stalled worker"
+            conflict_worker_id = ?conflict_worker_id,
+            "Forcefully taking over EOA lock from stalled worker."
         );
         // Force set - no expiry, only released by explicit takeover
         let _: () = conn.set(&lock_key, worker_id).await?;

executors/src/eoa/worker/confirm.rs

@@ -27,7 +27,7 @@ pub struct ConfirmedTransactionWithRichReceipt {
 
 impl<C: Chain> EoaExecutorWorker<C> {
     // ========== CONFIRM FLOW ==========
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(skip_all, fields(worker_id = self.store.worker_id))]
     pub async fn confirm_flow(&self) -> Result<CleanupReport, EoaExecutorWorkerError> {
         // Get fresh on-chain transaction count
         let current_chain_transaction_count = self
@@ -53,6 +53,10 @@ impl<C: Chain> EoaExecutorWorker<C> {
         let cached_transaction_count = match self.store.get_cached_transaction_count().await {
             Err(e) => match e {
                 TransactionStoreError::NonceSyncRequired { .. } => {
+                    tracing::warn!(
+                        cached_transaction_count = current_chain_transaction_count,
+                        "Nonce sync required, store was uninitialized, updating cached transaction count with current chain transaction count"
+                    );
                     self.store
                         .update_cached_transaction_count(current_chain_transaction_count)
                         .await?;

executors/src/eoa/worker/error.rs

@@ -140,8 +140,9 @@ impl UserCancellable for EoaExecutorWorkerError {
 // ========== SIMPLE ERROR CLASSIFICATION ==========
 #[derive(Debug)]
 pub enum SendErrorClassification {
-    PossiblySent,         // "nonce too low", "already known" etc
-    DeterministicFailure, // Invalid signature, malformed tx, insufficient funds etc
+    PossiblySent,                     // "nonce too low", "already known" etc
+    DeterministicFailure,             // Invalid signature, malformed tx, insufficient funds etc
+    DeterministicFailureNonRetryable, // Non-retryable deterministic failure
 }
 
 #[derive(PartialEq, Eq, Debug)]
@@ -185,11 +186,14 @@ pub fn classify_send_error(
     if error_str.contains("malformed")
         || error_str.contains("gas limit")
         || error_str.contains("intrinsic gas too low")
-        || error_str.contains("oversized")
     {
         return SendErrorClassification::DeterministicFailure;
     }
 
+    if error_str.contains("oversized") {
+        return SendErrorClassification::DeterministicFailureNonRetryable;
+    }
+
     tracing::warn!(
         "Unknown send error: {}. PLEASE REPORT FOR ADDING CORRECT CLASSIFICATION [NOTIFY]",
         error_str
@@ -305,6 +309,15 @@ impl SubmissionResult {
                             transaction: borrowed_transaction.clone().into(),
                         }
                     }
+                    SendErrorClassification::DeterministicFailureNonRetryable => SubmissionResult {
+                        result: SubmissionResultType::Fail(
+                            EoaExecutorWorkerError::TransactionSendError {
+                                message: format!("Transaction send failed: {rpc_error}"),
+                                inner_error: rpc_error.to_engine_error(chain),
+                            },
+                        ),
+                        transaction: borrowed_transaction.clone().into(),
+                    },
                 }
             }
         }

executors/src/eoa/worker/mod.rs

@@ -23,7 +23,9 @@ 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::metrics::{
+    calculate_duration_seconds, current_timestamp_ms, record_eoa_job_processing_time,
+};
 use crate::webhook::WebhookJobHandler;
 
 pub mod confirm;
@@ -148,20 +150,22 @@ where
             })
             .map_err_nack(Some(Duration::from_secs(10)), RequeuePosition::Last)?;
 
+        let worker_id = format!("{}:{}", uuid::Uuid::new_v4(), job.lease_token);
+
         // 2. CREATE SCOPED STORE (acquires lock)
         let scoped = EoaExecutorStore::new(
             self.redis.clone(),
             self.namespace.clone(),
             data.eoa_address,
             data.chain_id,
         )
-        .acquire_eoa_lock_aggressively(&job.lease_token)
+        .acquire_eoa_lock_aggressively(&worker_id)
         .await
         .map_err(|e| Into::<EoaExecutorWorkerError>::into(e).handle())?;
 
         let delegated_account = DelegatedAccount::new(data.eoa_address, chain.clone());
 
-        // if there's an error checking 7702 delegation here, we'll just assume it's not a minimal account for the pursposes of max in flight
+        // if there's an error checking 7702 delegation here, we'll just assume it's not a minimal account for the purposes of max in flight
         let is_minimal_account = self
             .authorization_cache
             .is_minimal_account(&delegated_account)
@@ -192,7 +196,7 @@ where
         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");
+            tracing::error!(error = ?e, worker_id = worker_id, "Error releasing EOA lock");
         }
 
         // Record EOA job processing metrics

executors/src/eoa/worker/send.rs

@@ -16,7 +16,7 @@ const HEALTH_CHECK_INTERVAL: u64 = 300; // 5 minutes in seconds
 
 impl<C: Chain> EoaExecutorWorker<C> {
     // ========== SEND FLOW ==========
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(skip_all, fields(worker_id = self.store.worker_id))]
     pub async fn send_flow(&self) -> Result<u32, EoaExecutorWorkerError> {
         // 1. Get EOA health (initializes if needed) and check if we should update balance
         let mut health = self.get_eoa_health().await?;