Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 47 additions & 32 deletions executors/src/eoa/store/atomic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -574,39 +574,54 @@ impl AtomicEoaExecutorStore {
error: EoaExecutorWorkerError,
webhook_queue: Arc<twmq::Queue<WebhookJobHandler>>,
) -> Result<(), TransactionStoreError> {
self.with_lock_check(|pipeline| {
let pending_key = self.pending_transactions_zset_name();
let tx_data_key = self.transaction_data_key_name(&pending_transaction.transaction_id);
let now = chrono::Utc::now().timestamp_millis().max(0) as u64;

// Remove from pending state
pipeline.zrem(&pending_key, &pending_transaction.transaction_id);

// Update transaction data with failure
pipeline.hset(&tx_data_key, "completed_at", now);
pipeline.hset(&tx_data_key, "failure_reason", error.to_string());
pipeline.hset(&tx_data_key, "status", "failed");

let event = EoaExecutorEvent {
transaction_id: pending_transaction.transaction_id.clone(),
address: pending_transaction.user_request.from,
};

let fail_envelope = event.transaction_failed_envelope(error.clone(), 1);

if !pending_transaction.user_request.webhook_options.is_empty() {
let mut tx_context = webhook_queue.transaction_context_from_pipeline(pipeline);
if let Err(e) = queue_webhook_envelopes(
fail_envelope,
pending_transaction.user_request.webhook_options.clone(),
&mut tx_context,
webhook_queue.clone(),
) {
tracing::error!("Failed to queue webhook for fail: {}", e);
}
let mut pipeline = twmq::redis::pipe();
pipeline.atomic();

let pending_key = self.pending_transactions_zset_name();
let tx_data_key = self.transaction_data_key_name(&pending_transaction.transaction_id);
let now = chrono::Utc::now().timestamp_millis().max(0) as u64;

// Remove from pending state
pipeline.zrem(&pending_key, &pending_transaction.transaction_id);

// Update transaction data with failure
pipeline.hset(&tx_data_key, "completed_at", now);
pipeline.hset(&tx_data_key, "failure_reason", error.to_string());
pipeline.hset(&tx_data_key, "status", "failed");

let event = EoaExecutorEvent {
transaction_id: pending_transaction.transaction_id.clone(),
address: pending_transaction.user_request.from,
};

let fail_envelope = event.transaction_failed_envelope(error.clone(), 1);

if !pending_transaction.user_request.webhook_options.is_empty() {
let mut tx_context = webhook_queue.transaction_context_from_pipeline(&mut pipeline);
if let Err(e) = queue_webhook_envelopes(
fail_envelope,
pending_transaction.user_request.webhook_options.clone(),
&mut tx_context,
webhook_queue.clone(),
) {
tracing::error!("Failed to queue webhook for fail: {}", e);
}
})
.await
}

let mut conn = self.redis.clone();
pipeline
.query_async::<Vec<twmq::redis::Value>>(&mut conn)
.await?;

tracing::info!(
transaction_id = %pending_transaction.transaction_id,
eoa = ?self.eoa(),
chain_id = self.chain_id(),
error = %error,
"JOB_LIFECYCLE - Deleted failed pending transaction from EOA"
);

Ok(())
}

pub async fn clean_submitted_transactions(
Expand Down