openzeppelin_relayer/jobs/handlers/

transaction_status_handler.rs

//! Transaction status monitoring handler.
//!
//! Monitors the status of submitted transactions by:
//! - Checking transaction status on the network
//! - Updating transaction status in storage
//! - Tracking failure counts for circuit breaker decisions (stored in transaction metadata)
use actix_web::web::ThinData;
use eyre::Result;
use tracing::{debug, info, instrument, warn};

use crate::{
    constants::{get_max_consecutive_status_failures, get_max_total_status_failures},
    domain::{get_relayer_transaction, get_transaction_by_id, is_final_state, Transaction},
    jobs::{Job, StatusCheckContext, TransactionStatusCheck},
    models::{
        ApiError, DefaultAppState, TransactionError, TransactionMetadata, TransactionRepoModel,
    },
    observability::request_id::set_request_id,
    queues::{HandlerError, WorkerContext},
    repositories::TransactionRepository,
};

#[instrument(
    level = "debug",
    skip(job, state, ctx),
    fields(
        request_id = ?job.request_id,
        job_id = %job.message_id,
        job_type = %job.job_type.to_string(),
        attempt = %ctx.attempt,
        tx_id = %job.data.transaction_id,
        relayer_id = %job.data.relayer_id,
        task_id = %ctx.task_id,
    )
)]
pub async fn transaction_status_handler(
    job: Job<TransactionStatusCheck>,
    state: ThinData<DefaultAppState>,
    ctx: WorkerContext,
) -> Result<(), HandlerError> {
    if let Some(request_id) = job.request_id.clone() {
        set_request_id(request_id);
    }

    let tx_repo = state.transaction_repository();

    // Execute status check - all logic moved here so errors go through handle_result
    let req_result = handle_request(&job.data, &state, ctx.attempt, &ctx.task_id).await;

    let tx_id = &job.data.transaction_id;

    // Handle result and update counters via transaction repository
    handle_result(
        req_result.result,
        &*tx_repo,
        tx_id,
        req_result.metadata,
        req_result.should_retry_on_error,
    )
    .await
}

/// Handles status check results with circuit breaker tracking.
///
/// # Strategy
/// - If transaction is in final state → return Ok (job completes, metadata cleaned up via delete_at)
/// - If success but not final → Reset consecutive to 0, return Err (retries)
/// - If error with should_retry=true → Increment counters, return Err (retries)
/// - If error with should_retry=false → Return Ok (job completes, e.g., transaction not found)
/// - If counters are None (early failure) → Skip counter updates
///
/// Counters are stored in transaction metadata, persisted via atomic Lua scripts.
async fn handle_result<TR>(
    result: Result<TransactionRepoModel>,
    tx_repo: &TR,
    tx_id: &str,
    metadata: Option<TransactionMetadata>,
    should_retry_on_error: bool,
) -> Result<(), HandlerError>
where
    TR: TransactionRepository + Send + Sync,
{
    match result {
        Ok(tx) if is_final_state(&tx.status) => {
            // Transaction reached final state - job complete
            // No need to clean up counters - tx will be deleted via delete_at
            debug!(
                tx_id = %tx.id,
                relayer_id = %tx.relayer_id,
                status = ?tx.status,
                consecutive_failures = ?metadata.as_ref().map(|m| m.consecutive_failures),
                total_failures = ?metadata.as_ref().map(|m| m.total_failures),
                "transaction in final state, status check complete"
            );

            Ok(())
        }
        Ok(tx) => {
            // Success but not final - RESET consecutive counter, keep total unchanged
            debug!(
                tx_id = %tx.id,
                relayer_id = %tx.relayer_id,
                status = ?tx.status,
                "transaction not in final state"
            );

            // Use fresh metadata from the transaction (updated during handle_transaction_status)
            // to decide whether a reset is needed, falling back to the pre-check snapshot.
            let fresh_meta = tx.metadata.clone().or(metadata);
            if let Some(meta) = fresh_meta {
                if meta.consecutive_failures > 0 {
                    if let Err(e) = tx_repo
                        .reset_status_check_consecutive_failures(tx_id.to_string())
                        .await
                    {
                        warn!(error = %e, tx_id = %tx_id, relayer_id = %tx.relayer_id, "failed to reset consecutive counter");
                    }
                }
            }

            // Return error to trigger retry
            Err(HandlerError::Retry(format!(
                "transaction status: {:?} - not in final state, retrying",
                tx.status
            )))
        }
        Err(e) => {
            if e.downcast_ref::<TransactionError>()
                .is_some_and(TransactionError::is_concurrent_update_conflict)
            {
                info!(
                    error = %e,
                    tx_id = %tx_id,
                    "status check lost a concurrent update race, completing job without counter changes"
                );
                return Ok(());
            }

            // Check if this is a permanent failure that shouldn't retry
            if !should_retry_on_error {
                info!(
                    error = %e,
                    tx_id = %tx_id,
                    "status check failed with permanent error, completing job without retry"
                );
                return Ok(());
            }

            // Transient error - INCREMENT both counters (only if we have metadata)
            if let Some(meta) = metadata {
                warn!(
                    error = %e,
                    tx_id = %tx_id,
                    consecutive_failures = meta.consecutive_failures.saturating_add(1),
                    total_failures = meta.total_failures.saturating_add(1),
                    "status check failed, incrementing failure counters"
                );

                // Update counters via atomic transaction repository method
                if let Err(update_err) = tx_repo
                    .increment_status_check_failures(tx_id.to_string())
                    .await
                {
                    warn!(error = %update_err, tx_id = %tx_id, "failed to update counters");
                }
            } else {
                // Early failure before counters were read - skip counter update
                warn!(
                    error = %e,
                    tx_id = %tx_id,
                    "status check failed early, counters not available"
                );
            }

            // Return error to trigger retry
            Err(HandlerError::Retry(format!("{e}")))
        }
    }
}

/// Result of handle_request including whether to retry on error.
struct HandleRequestResult {
    result: Result<TransactionRepoModel>,
    /// Transaction metadata with failure counters. None if metadata couldn't be read
    /// (e.g., transaction fetch failed early).
    metadata: Option<TransactionMetadata>,
    /// If false, errors should not trigger retry (e.g., transaction not found)
    should_retry_on_error: bool,
}

/// Executes the status check logic and returns the result with counter values.
/// Returns None for counters if they couldn't be read (e.g., transaction fetch failed early).
/// Sets should_retry_on_error=false for permanent failures like transaction not found.
async fn handle_request(
    status_request: &TransactionStatusCheck,
    state: &ThinData<DefaultAppState>,
    attempt: usize,
    task_id: &str,
) -> HandleRequestResult {
    let tx_id = &status_request.transaction_id;
    debug!(
        tx_id = %tx_id,
        relayer_id = %status_request.relayer_id,
        "handling transaction status check"
    );

    // Fetch transaction - if this fails, we can't read counters yet
    let transaction = match get_transaction_by_id(tx_id.clone(), state).await {
        Ok(tx) => tx,
        Err(ApiError::NotFound(msg)) => {
            // Transaction not found - permanent failure, don't retry
            warn!(tx_id = %tx_id, "transaction not found, completing job without retry: {}", msg);
            return HandleRequestResult {
                result: Err(eyre::eyre!("Transaction not found: {}", msg)),
                metadata: None,
                should_retry_on_error: false,
            };
        }
        Err(e) => {
            // Other errors - should retry
            return HandleRequestResult {
                result: Err(e.into()),
                metadata: None,
                should_retry_on_error: true,
            };
        }
    };

    // Read failure counters from transaction metadata
    let meta = transaction.metadata.clone().unwrap_or_default();

    // Get network type from transaction (authoritative source)
    let network_type = transaction.network_type;
    let max_consecutive = get_max_consecutive_status_failures(network_type);
    let max_total = get_max_total_status_failures(network_type);

    debug!(
        tx_id = %tx_id,
        consecutive_failures = meta.consecutive_failures,
        total_failures = meta.total_failures,
        max_consecutive,
        max_total,
        attempt,
        task_id = %task_id,
        "handling transaction status check"
    );

    // Build circuit breaker context
    let context = StatusCheckContext::new(
        meta.consecutive_failures,
        meta.total_failures,
        attempt as u32,
        max_consecutive,
        max_total,
        network_type,
    );

    // Get relayer transaction handler
    let relayer_transaction =
        match get_relayer_transaction(status_request.relayer_id.clone(), state).await {
            Ok(rt) => rt,
            Err(ApiError::NotFound(msg)) => {
                // Relayer or signer not found - permanent failure, don't retry
                warn!(
                    tx_id = %tx_id,
                    relayer_id = %status_request.relayer_id,
                    "relayer or signer not found, completing job without retry: {}", msg
                );
                return HandleRequestResult {
                    result: Err(eyre::eyre!("Relayer or signer not found: {}", msg)),
                    metadata: Some(meta),
                    should_retry_on_error: false,
                };
            }
            Err(e) => {
                // Other errors - should retry
                return HandleRequestResult {
                    result: Err(e.into()),
                    metadata: Some(meta),
                    should_retry_on_error: true,
                };
            }
        };

    // Execute status check
    let result = relayer_transaction
        .handle_transaction_status(transaction, Some(context))
        .await
        .map_err(|e| e.into());

    if let Ok(tx) = result.as_ref() {
        debug!(
            tx_id = %tx.id,
            status = ?tx.status,
            "status check handled successfully"
        );
    }

    HandleRequestResult {
        result,
        metadata: Some(meta),
        should_retry_on_error: true,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        models::{NetworkType, TransactionStatus},
        repositories::MockTransactionRepository,
    };
    use std::collections::HashMap;

    #[tokio::test]
    async fn test_status_check_job_validation() {
        let check_job = TransactionStatusCheck::new("tx123", "relayer-1", NetworkType::Evm);
        let job = Job::new(crate::jobs::JobType::TransactionStatusCheck, check_job);

        assert_eq!(job.data.transaction_id, "tx123");
        assert_eq!(job.data.relayer_id, "relayer-1");
        assert!(job.data.metadata.is_none());
    }

    #[tokio::test]
    async fn test_status_check_with_metadata() {
        let mut metadata = HashMap::new();
        metadata.insert("retry_count".to_string(), "2".to_string());
        metadata.insert("last_status".to_string(), "pending".to_string());

        let check_job = TransactionStatusCheck::new("tx123", "relayer-1", NetworkType::Evm)
            .with_metadata(metadata.clone());

        assert!(check_job.metadata.is_some());
        let job_metadata = check_job.metadata.unwrap();
        assert_eq!(job_metadata.get("retry_count").unwrap(), "2");
        assert_eq!(job_metadata.get("last_status").unwrap(), "pending");
    }

    #[test]
    fn test_status_check_network_type_required() {
        // Jobs should always have network_type set
        let check_job = TransactionStatusCheck::new("tx123", "relayer-1", NetworkType::Evm);
        assert!(check_job.network_type.is_some());

        // Verify different network types are preserved
        let solana_job = TransactionStatusCheck::new("tx456", "relayer-2", NetworkType::Solana);
        assert_eq!(solana_job.network_type, Some(NetworkType::Solana));

        let stellar_job = TransactionStatusCheck::new("tx789", "relayer-3", NetworkType::Stellar);
        assert_eq!(stellar_job.network_type, Some(NetworkType::Stellar));
    }

    mod context_tests {
        use super::*;

        #[test]
        fn test_context_should_force_finalize_below_threshold() {
            let ctx = StatusCheckContext::new(5, 10, 15, 25, 75, NetworkType::Evm);
            assert!(!ctx.should_force_finalize());
        }

        #[test]
        fn test_context_should_force_finalize_consecutive_at_threshold() {
            let ctx = StatusCheckContext::new(25, 30, 35, 25, 75, NetworkType::Evm);
            assert!(ctx.should_force_finalize());
        }

        #[test]
        fn test_context_should_force_finalize_total_at_threshold() {
            let ctx = StatusCheckContext::new(10, 75, 80, 25, 75, NetworkType::Evm);
            assert!(ctx.should_force_finalize());
        }
    }

    mod final_state_tests {
        use super::*;

        fn verify_final_state(status: TransactionStatus) {
            assert!(is_final_state(&status));
        }

        fn verify_not_final_state(status: TransactionStatus) {
            assert!(!is_final_state(&status));
        }

        #[test]
        fn test_confirmed_is_final() {
            verify_final_state(TransactionStatus::Confirmed);
        }

        #[test]
        fn test_failed_is_final() {
            verify_final_state(TransactionStatus::Failed);
        }

        #[test]
        fn test_canceled_is_final() {
            verify_final_state(TransactionStatus::Canceled);
        }

        #[test]
        fn test_expired_is_final() {
            verify_final_state(TransactionStatus::Expired);
        }

        #[test]
        fn test_pending_is_not_final() {
            verify_not_final_state(TransactionStatus::Pending);
        }

        #[test]
        fn test_sent_is_not_final() {
            verify_not_final_state(TransactionStatus::Sent);
        }

        #[test]
        fn test_submitted_is_not_final() {
            verify_not_final_state(TransactionStatus::Submitted);
        }

        #[test]
        fn test_mined_is_not_final() {
            verify_not_final_state(TransactionStatus::Mined);
        }
    }

    mod handle_result_tests {
        use super::*;

        /// Tests that counter increment uses saturating_add to prevent overflow
        #[test]
        fn test_counter_increment_saturating() {
            let consecutive: u32 = u32::MAX;
            let total: u32 = u32::MAX;

            let new_consecutive = consecutive.saturating_add(1);
            let new_total = total.saturating_add(1);

            // Should not overflow, stays at MAX
            assert_eq!(new_consecutive, u32::MAX);
            assert_eq!(new_total, u32::MAX);
        }

        /// Tests normal counter increment
        #[test]
        fn test_counter_increment_normal() {
            let consecutive: u32 = 5;
            let total: u32 = 10;

            let new_consecutive = consecutive.saturating_add(1);
            let new_total = total.saturating_add(1);

            assert_eq!(new_consecutive, 6);
            assert_eq!(new_total, 11);
        }

        /// Tests that consecutive counter resets to 0 on success (non-final)
        #[test]
        fn test_consecutive_reset_on_success() {
            // When status check succeeds but tx is not final,
            // consecutive should reset to 0, total stays unchanged
            let total: u32 = 20;

            // On success, consecutive resets
            let new_consecutive = 0;
            let new_total = total; // unchanged

            assert_eq!(new_consecutive, 0);
            assert_eq!(new_total, 20);
        }

        /// Tests that final states are correctly identified for cleanup
        #[test]
        fn test_final_state_triggers_cleanup() {
            let final_states = vec![
                TransactionStatus::Confirmed,
                TransactionStatus::Failed,
                TransactionStatus::Canceled,
                TransactionStatus::Expired,
            ];

            for status in final_states {
                assert!(
                    is_final_state(&status),
                    "Expected {status:?} to be a final state"
                );
            }
        }

        /// Tests that non-final states trigger retry
        #[test]
        fn test_non_final_state_triggers_retry() {
            let non_final_states = vec![
                TransactionStatus::Pending,
                TransactionStatus::Sent,
                TransactionStatus::Submitted,
                TransactionStatus::Mined,
            ];

            for status in non_final_states {
                assert!(
                    !is_final_state(&status),
                    "Expected {status:?} to NOT be a final state"
                );
            }
        }
    }

    mod handle_request_result_tests {
        use super::*;

        #[tokio::test]
        async fn test_handle_result_ignores_concurrent_update_conflict() {
            let tx_repo = MockTransactionRepository::new();

            let result = handle_result(
                Err(TransactionError::ConcurrentUpdateConflict("tx race".to_string()).into()),
                &tx_repo,
                "tx-1",
                Some(TransactionMetadata {
                    consecutive_failures: 2,
                    total_failures: 5,
                    ..Default::default()
                }),
                true,
            )
            .await;

            assert!(result.is_ok());
        }

        #[test]
        fn test_handle_request_result_with_counters() {
            let result = HandleRequestResult {
                result: Ok(TransactionRepoModel::default()),
                metadata: Some(TransactionMetadata {
                    consecutive_failures: 5,
                    total_failures: 10,
                    insufficient_fee_retries: 2,
                    try_again_later_retries: 1,
                }),
                should_retry_on_error: true,
            };

            assert!(result.result.is_ok());
            let meta = result.metadata.unwrap();
            assert_eq!(meta.consecutive_failures, 5);
            assert_eq!(meta.total_failures, 10);
            assert!(result.should_retry_on_error);
        }

        #[test]
        fn test_handle_request_result_without_counters() {
            // Early failure before counters could be read
            let result = HandleRequestResult {
                result: Err(eyre::eyre!("Transaction not found")),
                metadata: None,
                should_retry_on_error: false,
            };

            assert!(result.result.is_err());
            assert!(result.metadata.is_none());
            assert!(!result.should_retry_on_error);
        }

        #[test]
        fn test_permanent_error_should_not_retry() {
            // NotFound errors are permanent - should not retry
            let result = HandleRequestResult {
                result: Err(eyre::eyre!("Transaction not found")),
                metadata: None,
                should_retry_on_error: false,
            };

            // Permanent errors have should_retry_on_error = false
            assert!(!result.should_retry_on_error);
        }

        #[test]
        fn test_transient_error_should_retry() {
            // Network/connection errors are transient - should retry
            let result = HandleRequestResult {
                result: Err(eyre::eyre!("Connection timeout")),
                metadata: Some(TransactionMetadata {
                    consecutive_failures: 3,
                    total_failures: 7,
                    insufficient_fee_retries: 1,
                    try_again_later_retries: 0,
                }),
                should_retry_on_error: true,
            };

            // Transient errors have should_retry_on_error = true
            assert!(result.should_retry_on_error);
        }
    }
}