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, TransactionMetadata, TransactionRepoModel,
        TransactionUpdateRequest,
    },
    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.consecutive_failures,
        req_result.total_failures,
        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 partial_update.
async fn handle_result<TR>(
    result: Result<TransactionRepoModel>,
    tx_repo: &TR,
    tx_id: &str,
    consecutive_failures: Option<u32>,
    total_failures: Option<u32>,
    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 = ?consecutive_failures,
                total_failures = ?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"
            );

            // Reset consecutive counter only if there were previous failures
            // This avoids unnecessary writes for transactions that never failed
            match (consecutive_failures, total_failures) {
                (Some(consecutive), Some(total)) if consecutive > 0 || total > 0 => {
                    let update = TransactionUpdateRequest {
                        metadata: Some(TransactionMetadata {
                            consecutive_failures: 0,
                            total_failures: total,
                        }),
                        ..Default::default()
                    };
                    if let Err(e) = tx_repo.partial_update(tx_id.to_string(), update).await {
                        warn!(error = %e, tx_id = %tx_id, relayer_id = %tx.relayer_id, "failed to reset consecutive counter");
                    }
                }
                _ => {
                    // No previous failures or counters not available - nothing to reset
                }
            }

            // Return error to trigger retry
            Err(HandlerError::Retry(format!(
                "transaction status: {:?} - not in final state, retrying",
                tx.status
            )))
        }
        Err(e) => {
            // 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 values)
            match (consecutive_failures, total_failures) {
                (Some(consecutive), Some(total)) => {
                    let new_consecutive = consecutive.saturating_add(1);
                    let new_total = total.saturating_add(1);

                    warn!(
                        error = %e,
                        tx_id = %tx_id,
                        consecutive_failures = new_consecutive,
                        total_failures = new_total,
                        "status check failed, incrementing failure counters"
                    );

                    // Update counters via transaction repository
                    let update = TransactionUpdateRequest {
                        metadata: Some(TransactionMetadata {
                            consecutive_failures: new_consecutive,
                            total_failures: new_total,
                        }),
                        ..Default::default()
                    };
                    if let Err(update_err) = tx_repo.partial_update(tx_id.to_string(), update).await
                    {
                        warn!(error = %update_err, tx_id = %tx_id, "failed to update counters");
                    }
                }
                _ => {
                    // 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>,
    consecutive_failures: Option<u32>,
    total_failures: Option<u32>,
    /// 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)),
                consecutive_failures: None,
                total_failures: None,
                should_retry_on_error: false,
            };
        }
        Err(e) => {
            // Other errors - should retry
            return HandleRequestResult {
                result: Err(e.into()),
                consecutive_failures: None,
                total_failures: None,
                should_retry_on_error: true,
            };
        }
    };

    // Read failure counters from transaction metadata
    let (consecutive_failures, total_failures) = match &transaction.metadata {
        Some(meta) => (meta.consecutive_failures, meta.total_failures),
        None => (0, 0),
    };

    // 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,
        total_failures,
        max_consecutive,
        max_total,
        attempt,
        task_id = %task_id,
        "handling transaction status check"
    );

    // Build circuit breaker context
    let context = StatusCheckContext::new(
        consecutive_failures,
        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)),
                    consecutive_failures: Some(consecutive_failures),
                    total_failures: Some(total_failures),
                    should_retry_on_error: false,
                };
            }
            Err(e) => {
                // Other errors - should retry
                return HandleRequestResult {
                    result: Err(e.into()),
                    consecutive_failures: Some(consecutive_failures),
                    total_failures: Some(total_failures),
                    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,
        consecutive_failures: Some(consecutive_failures),
        total_failures: Some(total_failures),
        should_retry_on_error: true,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::{NetworkType, TransactionStatus};
    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::*;

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

            assert!(result.result.is_ok());
            assert_eq!(result.consecutive_failures, Some(5));
            assert_eq!(result.total_failures, Some(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")),
                consecutive_failures: None,
                total_failures: None,
                should_retry_on_error: false,
            };

            assert!(result.result.is_err());
            assert_eq!(result.consecutive_failures, None);
            assert_eq!(result.total_failures, 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")),
                consecutive_failures: None,
                total_failures: 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")),
                consecutive_failures: Some(3),
                total_failures: Some(7),
                should_retry_on_error: true,
            };

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