hypercall_transaction_submitter/

lib.rs

pub mod exchange_encoder;
mod tx_relayer;
pub use exchange_encoder::{encode_execute_rsm_directive, encode_execute_user_directive};
pub use tx_relayer::{
    AwsKmsTxRelayer, DirectTxRelayer, MockTxRelayer, PendingTransactionStatus,
    SubmittedTransaction, TxRelayerTrait,
};
use tx_relayer::{SignedAttemptRecorder, SignedTransactionAttempt};

use alloy::primitives::{Address, Bytes};
use async_trait::async_trait;
use eyre::Result as EyreResult;
use hypercall_db::AsyncDirectiveOutboxWriter;
use hypercall_transaction_submitter_core::{
    ContractCall, SubmissionStatus, SubmittedNonce, SubmitterId,
};
use hypercall_transaction_submitter_db::{
    AsyncTransactionSubmitterStore, SubmissionAttemptRecord, SubmissionRecord,
};
use hypercall_types::{
    topics::TOPIC_TRANSACTION_REQUESTS, EngineMessage, SignedDirectiveTx, TransactionRequest,
    TransactionStatus, TransactionType, TransactionUpdate,
};
use serde::Deserialize;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{mpsc, Mutex};
use tracing::{debug, error, info};

#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum TransactionSubmitterMode {
    Mock,
    Direct,
    AwsKms,
}

impl FromStr for TransactionSubmitterMode {
    type Err = String;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value.trim().to_ascii_lowercase().as_str() {
            "mock" => Ok(Self::Mock),
            "direct" => Ok(Self::Direct),
            "aws_kms" => Ok(Self::AwsKms),
            other => Err(format!(
                "expected one of: mock, direct, aws_kms; got {}",
                other
            )),
        }
    }
}

#[derive(Debug, Clone, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct TransactionSubmitterConfig {
    pub mode: TransactionSubmitterMode,
    pub aws_kms_key_id: String,
    pub max_gas_price: String,
    pub rpc_url: String,
}

impl Default for TransactionSubmitterConfig {
    fn default() -> Self {
        Self {
            mode: TransactionSubmitterMode::Mock,
            aws_kms_key_id: String::new(),
            max_gas_price: "500000000000".to_string(),
            rpc_url: "https://rpc.hyperliquid-testnet.xyz/evm".to_string(),
        }
    }
}

#[derive(Debug, Clone, Default)]
pub struct TransactionSubmitterSecrets {
    pub transaction_submitter_private_key: Option<String>,
}

impl TransactionSubmitterSecrets {
    pub fn require_transaction_submitter_private_key(&self) -> eyre::Result<&str> {
        self.transaction_submitter_private_key
            .as_deref()
            .ok_or_else(|| {
                eyre::eyre!("TRANSACTION_SUBMITTER_PRIVATE_KEY must be set in the environment")
            })
    }
}

#[async_trait]
pub trait EventBusTrait: Send + Sync {
    fn get_sender(&self) -> mpsc::UnboundedSender<EngineMessage>;

    async fn subscribe(
        &self,
        topics: Vec<String>,
    ) -> Result<mpsc::UnboundedReceiver<EngineMessage>, String>;
}

type DirectiveEncoder = fn(Address, Bytes, Bytes, Option<String>) -> EyreResult<ContractCall>;
type DirectiveOutboxDb = Arc<dyn AsyncDirectiveOutboxWriter>;
type SubmitterStoreDb = Arc<dyn AsyncTransactionSubmitterStore>;

const PENDING_RECONCILIATION_RETRY_DELAY: Duration = Duration::from_secs(30);
const PERSISTED_PENDING_SUBMISSION_BATCH_SIZE: i64 = 1000;

#[derive(Debug)]
enum RelayDirectiveError {
    Build(String),
    Relay(String),
    Terminal(String),
}

struct SubmitterAttemptRecorder {
    store: Option<SubmitterStoreDb>,
}

#[async_trait]
impl SignedAttemptRecorder for SubmitterAttemptRecorder {
    async fn record_signed_attempt(&self, attempt: SignedTransactionAttempt) -> EyreResult<()> {
        let Some(store) = self.store.as_ref() else {
            return Ok(());
        };
        let record = SubmissionRecord {
            submitter: attempt.submitter,
            nonce: attempt.nonce,
            status: SubmissionStatus::Created,
            primary_tx_hash: Some(attempt.tx_hash.clone()),
            terminal_error: None,
        };
        let attempts = vec![SubmissionAttemptRecord {
            tx_hash: attempt.tx_hash,
            raw_tx: attempt.raw_tx,
        }];
        TransactionSubmitter::persist_submitter_submission_with_db(
            Arc::clone(store),
            record,
            attempts,
        )
        .await
    }
}

/// Delivers signed directive requests to chain and publishes delivery status.
///
/// Signer-specific crates own key custody and signer construction. This type
/// owns the operational path around those signers: encode, submit, reconcile,
/// publish status, and persist outbox delivery updates.
pub struct TransactionSubmitter {
    event_bus: Arc<dyn EventBusTrait>,
    tx_relayer: Arc<dyn TxRelayerTrait>,
    exchange_address: Address,
    /// Optional direct DB handle for outbox delivery status updates.
    ///
    /// This keeps direct-submission deployments from relying only on the
    /// in-process event bus to record terminal delivery state.
    db: Option<DirectiveOutboxDb>,
    /// Optional submitter-owned persistence for submissions and replacement
    /// attempts.
    submitter_store: Option<SubmitterStoreDb>,
    /// Serializes nonce allocation and the first submit attempt for this
    /// submitter instance.
    ///
    /// This mirrors the Sanguine submitter invariant: no later request may
    /// observe or allocate around a nonce while an earlier request is still in
    /// the undecided window between nonce selection and durable submitter state.
    nonce_lock: Arc<Mutex<()>>,
}

impl TransactionSubmitter {
    /// Creates TransactionSubmitter with direct RPC submission using the configured signer key.
    pub async fn new_direct(
        event_bus: Arc<dyn EventBusTrait>,
        config: &TransactionSubmitterConfig,
        exchange_contract_address: &str,
        secrets: &TransactionSubmitterSecrets,
    ) -> EyreResult<Self> {
        let tx_relayer = DirectTxRelayer::new(config, secrets).await?;
        let exchange_address = exchange_encoder::parse_exchange_address(exchange_contract_address)?;
        Ok(Self {
            event_bus,
            tx_relayer: Arc::new(tx_relayer),
            exchange_address,
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        })
    }

    /// Creates TransactionSubmitter with direct RPC submission using AWS KMS.
    pub async fn new_aws_kms(
        event_bus: Arc<dyn EventBusTrait>,
        config: &TransactionSubmitterConfig,
        exchange_contract_address: &str,
    ) -> EyreResult<Self> {
        let tx_relayer = AwsKmsTxRelayer::new(config).await?;
        let exchange_address = exchange_encoder::parse_exchange_address(exchange_contract_address)?;
        Ok(Self {
            event_bus,
            tx_relayer: Arc::new(tx_relayer),
            exchange_address,
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        })
    }

    /// Creates TransactionSubmitter with no-op mock for local tests and disabled delivery.
    pub fn new_mock(
        event_bus: Arc<dyn EventBusTrait>,
        exchange_contract_address: &str,
    ) -> EyreResult<Self> {
        Ok(Self {
            event_bus,
            tx_relayer: Arc::new(MockTxRelayer),
            exchange_address: exchange_encoder::parse_exchange_address(exchange_contract_address)?,
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        })
    }

    /// Attach a database handler for direct outbox persistence.
    /// When set, outbox updates are written to Postgres directly,
    /// independent of the event bus consumer.
    pub fn with_db(mut self, db: DirectiveOutboxDb) -> Self {
        self.db = Some(db);
        self
    }

    pub fn with_submitter_store(mut self, submitter_store: SubmitterStoreDb) -> Self {
        self.submitter_store = Some(submitter_store);
        self
    }

    /// Start the transaction submission service (legacy, no shutdown support).
    /// For graceful shutdown support, use `start_with_shutdown`.
    pub fn start(self: Arc<Self>) -> Result<tokio::task::JoinHandle<()>, String> {
        // Create a dummy shutdown channel that never fires.
        // IMPORTANT: We must keep the sender alive, otherwise recv() returns Closed immediately.
        let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
        std::mem::forget(tx);
        self.start_with_shutdown(rx)
    }

    /// Start the transaction submission service with shutdown support.
    ///
    /// Returns a `JoinHandle` for integration with `TaskGroup` for coordinated shutdown.
    pub fn start_with_shutdown(
        self: Arc<Self>,
        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<tokio::task::JoinHandle<()>, String> {
        info!("Starting Transaction Submitter service...");

        // Subscribe to transaction requests
        let event_bus = self.event_bus.clone();
        let submitter = self.clone();
        let handle = tokio::spawn(async move {
            submitter.spawn_persisted_pending_submissions().await;

            let mut tx_receiver = match event_bus
                .subscribe(vec![TOPIC_TRANSACTION_REQUESTS.to_string()])
                .await
            {
                Ok(rx) => rx,
                Err(e) => {
                    error!("Failed to subscribe to transaction requests: {}", e);
                    return;
                }
            };

            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("Transaction Submitter received shutdown signal");
                        break;
                    }
                    maybe_message = tx_receiver.recv() => {
                        match maybe_message {
                            Some(EngineMessage::TransactionRequest(tx_req)) => {
                                submitter.process_transaction_request(tx_req).await;
                            }
                            Some(_) => {} // Ignore other message types
                            None => break, // Channel closed
                        }
                    }
                }
            }
            info!("Transaction Submitter stopped");
        });

        info!("✓ Transaction Submitter service started");
        Ok(handle)
    }

    async fn spawn_persisted_pending_submissions(&self) {
        let Some(submitter_store) = self.submitter_store.as_ref() else {
            return;
        };

        let mut after_submission_id = 0_i64;
        loop {
            let rows = match submitter_store
                .list_pending_submissions(
                    after_submission_id,
                    PERSISTED_PENDING_SUBMISSION_BATCH_SIZE,
                )
                .await
            {
                Ok(rows) => rows,
                Err(error) => {
                    error!("Failed to load persisted pending submitter submissions: {error}");
                    return;
                }
            };

            if rows.is_empty() {
                break;
            }

            for row in rows {
                after_submission_id = row.submission_id;
                if let Some(directive_id) = row.directive_id {
                    self.spawn_pending_reconciliation(
                        directive_id,
                        row.submitter,
                        row.nonce,
                        row.tx_hashes,
                    );
                } else {
                    self.spawn_pending_submitter_reconciliation(
                        row.submitter,
                        row.nonce,
                        row.tx_hashes,
                    );
                }
            }
        }
    }

    /// Process a single transaction request
    async fn process_transaction_request(&self, tx_req: TransactionRequest) {
        info!("Processing transaction request: {}", tx_req.request_id);

        // Check if request has expired
        let current_time = timestamp_millis();
        if current_time > tx_req.expires_at {
            self.persist_outbox_update(
                &tx_req.request_id,
                TransactionStatus::Expired,
                None,
                Some("Transaction request expired".to_string()),
            )
            .await;
            self.send_update(
                &tx_req.request_id,
                TransactionStatus::Expired,
                None,
                Some("Transaction request expired".to_string()),
            )
            .await;
            return;
        }

        let (directive, encoder, log_label): (&SignedDirectiveTx, DirectiveEncoder, &'static str) =
            match &tx_req.transaction_type {
                TransactionType::UserDirective(user_directive) => {
                    (user_directive, encode_execute_user_directive, "user")
                }
                TransactionType::RsmDirective(rsm_directive) => {
                    (rsm_directive, encode_execute_rsm_directive, "RSM")
                }
            };

        self.handle_signed_directive(
            &tx_req.request_id,
            &tx_req.account_contract,
            directive,
            encoder,
            log_label,
        )
        .await;
    }

    /// Orchestrate signed directive submission with status updates.
    async fn handle_signed_directive(
        &self,
        request_id: &str,
        account_contract: &hypercall_types::WalletAddress,
        directive: &SignedDirectiveTx,
        encoder: DirectiveEncoder,
        log_label: &str,
    ) {
        info!(
            "Submitting {} directive: request_id={}, account={}",
            log_label, request_id, account_contract
        );

        match self
            .submit_signed_directive(request_id, directive, encoder, log_label)
            .await
        {
            Ok(submission) => {
                let tx_hash = Some(submission.hash.clone());
                let status = if submission.confirmed {
                    TransactionStatus::Confirmed
                } else {
                    TransactionStatus::Pending
                };
                self.persist_outbox_update(request_id, status, tx_hash.clone(), None)
                    .await;
                self.send_update(request_id, status, tx_hash, None).await;
                if !submission.confirmed {
                    self.spawn_pending_reconciliation(
                        request_id.to_string(),
                        submission.submitter,
                        submission.nonce,
                        submission.all_hashes,
                    );
                }
                info!(
                    "{} directive submitted: {} confirmed={}",
                    log_label, request_id, submission.confirmed
                );
            }
            Err(RelayDirectiveError::Build(error)) => {
                error!(
                    "{} directive relay payload build failed: {} - {}; emitting terminal failure",
                    log_label, request_id, error
                );
                self.persist_outbox_update(
                    request_id,
                    TransactionStatus::Failed,
                    None,
                    Some(error.clone()),
                )
                .await;
                self.send_update(request_id, TransactionStatus::Failed, None, Some(error))
                    .await;
            }
            Err(RelayDirectiveError::Relay(error)) => {
                error!(
                    "{} directive relay submission failed: {} - {}; outbox row remains retryable",
                    log_label, request_id, error
                );
                self.persist_outbox_update(
                    request_id,
                    TransactionStatus::Submitted,
                    None,
                    Some(error.clone()),
                )
                .await;
                self.send_update(request_id, TransactionStatus::Submitted, None, Some(error))
                    .await;
            }
            Err(RelayDirectiveError::Terminal(error)) => {
                error!(
                    "{} directive relay terminal failure: {} - {}; emitting terminal failure",
                    log_label, request_id, error
                );
                self.persist_outbox_update(
                    request_id,
                    TransactionStatus::Failed,
                    None,
                    Some(error.clone()),
                )
                .await;
                self.send_update(request_id, TransactionStatus::Failed, None, Some(error))
                    .await;
            }
        }
    }

    /// Build a submitter-owned contract call from SignedDirectiveTx and submit it.
    async fn submit_signed_directive(
        &self,
        request_id: &str,
        signed_directive: &SignedDirectiveTx,
        encoder: DirectiveEncoder,
        log_label: &str,
    ) -> Result<SubmittedTransaction, RelayDirectiveError> {
        info!(
            "Building {} directive transaction payload: request_id={}",
            log_label, request_id
        );

        let directive = Bytes::from(signed_directive.directive.clone());
        let signature = Bytes::from_str(&signed_directive.signature)
            .map_err(|e| RelayDirectiveError::Build(format!("invalid signature bytes: {}", e)))?;
        let contract_call = encoder(
            self.exchange_address,
            directive,
            signature,
            Some(request_id.to_string()),
        )
        .map_err(|e| RelayDirectiveError::Build(e.to_string()))?;

        let calldata_hex = hex::encode(contract_call.data.as_ref());
        error!(
            request_id,
            log_label,
            exchange_address = %self.exchange_address,
            tx_to = %contract_call.to,
            tx_value = %contract_call.value,
            calldata = %format!("0x{calldata_hex}"),
            directive = %format!("0x{}", hex::encode(&signed_directive.directive)),
            signature = %signed_directive.signature,
            "prepared directive transaction payload"
        );

        self.submit_contract_call(request_id, contract_call)
            .await
            .map_err(|e| {
                let error = e.to_string();
                if is_terminal_relay_error(&error) {
                    RelayDirectiveError::Terminal(error)
                } else {
                    RelayDirectiveError::Relay(error)
                }
            })
    }

    async fn submit_contract_call(
        &self,
        request_id: &str,
        contract_call: ContractCall,
    ) -> EyreResult<SubmittedTransaction> {
        let _nonce_guard = self.nonce_lock.lock().await;
        let submitter = self.tx_relayer.submitter_address();
        let db_nonce_floor = match self.submitter_store.as_ref() {
            Some(store) => store
                .max_nonce_for_submitter(&submitter)
                .await
                .map_err(|error| eyre::eyre!("failed to load submitter nonce floor: {error}"))?
                .map(|nonce| nonce.saturating_add(1)),
            None => None,
        };
        let nonce = self.tx_relayer.select_next_nonce(db_nonce_floor).await?;
        self.persist_created_submitter_submission(submitter, nonce)
            .await?;
        self.persist_directive_submitter_pointer(request_id, submitter, nonce)
            .await?;

        let recorder: Arc<dyn SignedAttemptRecorder> = Arc::new(SubmitterAttemptRecorder {
            store: self.submitter_store.as_ref().map(Arc::clone),
        });
        let submission = match self
            .tx_relayer
            .send_transaction_with_nonce_recording_attempts(contract_call, nonce, recorder)
            .await
        {
            Ok(submission) => submission,
            Err(error) => {
                if let Some(submitter_store) = self.submitter_store.as_ref() {
                    Self::persist_submitter_status_with_db(
                        Arc::clone(submitter_store),
                        submitter,
                        nonce,
                        SubmissionStatus::Failed,
                        None,
                        Some(error.to_string()),
                    )
                    .await;
                }
                return Err(error);
            }
        };
        let status = if submission.confirmed {
            SubmissionStatus::Confirmed
        } else {
            SubmissionStatus::Broadcasted
        };
        if let Some(submitter_store) = self.submitter_store.as_ref() {
            Self::persist_submitter_status_with_db(
                Arc::clone(submitter_store),
                submission.submitter,
                submission.nonce,
                status,
                Some(submission.hash.clone()),
                None,
            )
            .await;
        }

        info!("Transaction submitted: {:?}", submission);
        Ok(submission)
    }

    /// Send transaction status update
    async fn send_update(
        &self,
        request_id: &str,
        status: TransactionStatus,
        tx_hash: Option<String>,
        error: Option<String>,
    ) {
        let update = TransactionUpdate {
            request_id: request_id.to_string(),
            status,
            tx_hash,
            error,
            timestamp: timestamp_millis(),
            gas_used: None,
            gas_price: None,
        };

        debug!("Sending transaction update: {:?}", update);

        if let Err(e) = self
            .event_bus
            .get_sender()
            .send(EngineMessage::TransactionUpdate(update))
        {
            error!("Failed to send transaction update: {}", e);
        }
    }

    /// Persist outbox delivery status directly through the DB writer interface.
    async fn persist_outbox_update(
        &self,
        request_id: &str,
        status: TransactionStatus,
        tx_hash: Option<String>,
        error_msg: Option<String>,
    ) {
        let Some(db) = self.db.as_ref() else {
            return;
        };
        Self::persist_outbox_update_with_db(
            Arc::clone(db),
            request_id.to_string(),
            status,
            tx_hash,
            error_msg,
        )
        .await;
    }

    async fn persist_outbox_update_with_db(
        db: DirectiveOutboxDb,
        request_id: String,
        status: TransactionStatus,
        tx_hash: Option<String>,
        error_msg: Option<String>,
    ) {
        if let Err(e) = db
            .persist_directive_transaction_update(
                &request_id,
                status,
                tx_hash.as_deref(),
                error_msg.as_deref(),
            )
            .await
        {
            error!("Failed to persist outbox update for {}: {}", request_id, e);
        }
    }

    async fn persist_created_submitter_submission(
        &self,
        submitter: SubmitterId,
        nonce: SubmittedNonce,
    ) -> EyreResult<()> {
        let Some(submitter_store) = self.submitter_store.as_ref() else {
            return Ok(());
        };
        let record = SubmissionRecord {
            submitter,
            nonce,
            status: SubmissionStatus::Created,
            primary_tx_hash: None,
            terminal_error: None,
        };
        Self::persist_submitter_submission_with_db(Arc::clone(submitter_store), record, Vec::new())
            .await
    }

    async fn persist_submitter_submission_with_db(
        submitter_store: SubmitterStoreDb,
        record: SubmissionRecord,
        attempts: Vec<SubmissionAttemptRecord>,
    ) -> EyreResult<()> {
        submitter_store
            .record_submission(&record, &attempts)
            .await
            .map_err(|e| {
                eyre::eyre!(
                    "failed to persist submitter submission {}:{}: {}",
                    record.submitter,
                    record.nonce,
                    e
                )
            })
    }

    async fn persist_submitter_status_with_db(
        submitter_store: SubmitterStoreDb,
        submitter: SubmitterId,
        nonce: SubmittedNonce,
        status: SubmissionStatus,
        primary_tx_hash: Option<String>,
        terminal_error: Option<String>,
    ) {
        if let Err(e) = submitter_store
            .update_submission_status(
                &submitter,
                nonce,
                status,
                primary_tx_hash.as_deref(),
                terminal_error.as_deref(),
            )
            .await
        {
            error!(
                "Failed to update submitter submission {}:{}: {}",
                submitter, nonce, e
            );
        }
    }

    async fn persist_directive_submitter_pointer(
        &self,
        request_id: &str,
        submitter: SubmitterId,
        nonce: SubmittedNonce,
    ) -> EyreResult<()> {
        let Some(db) = self.db.as_ref() else {
            return Ok(());
        };
        Self::persist_directive_submitter_pointer_with_db(
            Arc::clone(db),
            request_id.to_string(),
            submitter,
            nonce,
        )
        .await
    }

    async fn persist_directive_submitter_pointer_with_db(
        db: DirectiveOutboxDb,
        request_id: String,
        submitter: SubmitterId,
        nonce: SubmittedNonce,
    ) -> EyreResult<()> {
        db.record_directive_submitter_submission(&request_id, &submitter, nonce)
            .await
            .map_err(|e| {
                eyre::eyre!(
                    "failed to persist directive submitter pointer for {}: {}",
                    request_id,
                    e
                )
            })
    }

    fn spawn_pending_reconciliation(
        &self,
        request_id: String,
        submitter: SubmitterId,
        nonce: SubmittedNonce,
        tx_hashes: Vec<String>,
    ) {
        let sender = self.event_bus.get_sender();
        let tx_relayer = Arc::clone(&self.tx_relayer);
        let db = self.db.clone();
        let submitter_store = self.submitter_store.clone();

        tokio::spawn(async move {
            let primary_tx_hash = tx_hashes
                .last()
                .cloned()
                .unwrap_or_else(|| "unknown".to_string());
            loop {
                match tx_relayer.reconcile_pending_transactions(&tx_hashes).await {
                    Ok(PendingTransactionStatus::Unsupported) => {}
                    Ok(PendingTransactionStatus::Pending(reason)) => {
                        send_update_via_sender(
                            &sender,
                            request_id.clone(),
                            TransactionStatus::Pending,
                            Some(primary_tx_hash.clone()),
                            Some(reason),
                        );
                        tokio::time::sleep(PENDING_RECONCILIATION_RETRY_DELAY).await;
                        continue;
                    }
                    Ok(PendingTransactionStatus::Confirmed(confirmed_tx_hash)) => {
                        if let Some(submitter_store) = submitter_store.as_ref() {
                            Self::persist_submitter_status_with_db(
                                Arc::clone(submitter_store),
                                submitter,
                                nonce,
                                SubmissionStatus::Confirmed,
                                Some(confirmed_tx_hash.clone()),
                                None,
                            )
                            .await;
                        }
                        if let Some(db) = db.as_ref() {
                            Self::persist_outbox_update_with_db(
                                Arc::clone(db),
                                request_id.clone(),
                                TransactionStatus::Confirmed,
                                Some(confirmed_tx_hash.clone()),
                                None,
                            )
                            .await;
                        }
                        send_update_via_sender(
                            &sender,
                            request_id.clone(),
                            TransactionStatus::Confirmed,
                            Some(confirmed_tx_hash),
                            None,
                        );
                    }
                    Ok(PendingTransactionStatus::Failed(error)) => {
                        if let Some(submitter_store) = submitter_store.as_ref() {
                            Self::persist_submitter_status_with_db(
                                Arc::clone(submitter_store),
                                submitter,
                                nonce,
                                SubmissionStatus::Failed,
                                Some(primary_tx_hash.clone()),
                                Some(error.clone()),
                            )
                            .await;
                        }
                        if let Some(db) = db.as_ref() {
                            Self::persist_outbox_update_with_db(
                                Arc::clone(db),
                                request_id.clone(),
                                TransactionStatus::Failed,
                                Some(primary_tx_hash.clone()),
                                Some(error.clone()),
                            )
                            .await;
                        }
                        send_update_via_sender(
                            &sender,
                            request_id.clone(),
                            TransactionStatus::Failed,
                            Some(primary_tx_hash.clone()),
                            Some(error),
                        );
                    }
                    Err(error) => {
                        send_update_via_sender(
                            &sender,
                            request_id.clone(),
                            TransactionStatus::Pending,
                            Some(primary_tx_hash.clone()),
                            Some(format!(
                                "pending transaction reconciliation failed: {:?}",
                                error
                            )),
                        );
                        tokio::time::sleep(PENDING_RECONCILIATION_RETRY_DELAY).await;
                        continue;
                    }
                }
                break;
            }
        });
    }

    fn spawn_pending_submitter_reconciliation(
        &self,
        submitter: SubmitterId,
        nonce: SubmittedNonce,
        tx_hashes: Vec<String>,
    ) {
        let tx_relayer = Arc::clone(&self.tx_relayer);
        let submitter_store = self.submitter_store.clone();

        tokio::spawn(async move {
            loop {
                match tx_relayer.reconcile_pending_transactions(&tx_hashes).await {
                    Ok(PendingTransactionStatus::Unsupported) => {}
                    Ok(PendingTransactionStatus::Pending(reason)) => {
                        tracing::debug!(
                            %submitter,
                            %nonce,
                            %reason,
                            "submitter submission still pending"
                        );
                        tokio::time::sleep(PENDING_RECONCILIATION_RETRY_DELAY).await;
                        continue;
                    }
                    Ok(PendingTransactionStatus::Confirmed(confirmed_tx_hash)) => {
                        if let Some(submitter_store) = submitter_store.as_ref() {
                            Self::persist_submitter_status_with_db(
                                Arc::clone(submitter_store),
                                submitter,
                                nonce,
                                SubmissionStatus::Confirmed,
                                Some(confirmed_tx_hash),
                                None,
                            )
                            .await;
                        }
                    }
                    Ok(PendingTransactionStatus::Failed(error)) => {
                        if let Some(submitter_store) = submitter_store.as_ref() {
                            Self::persist_submitter_status_with_db(
                                Arc::clone(submitter_store),
                                submitter,
                                nonce,
                                SubmissionStatus::Failed,
                                None,
                                Some(error),
                            )
                            .await;
                        }
                    }
                    Err(error) => {
                        tracing::warn!(
                            %submitter,
                            %nonce,
                            ?error,
                            "submitter pending reconciliation failed"
                        );
                        tokio::time::sleep(PENDING_RECONCILIATION_RETRY_DELAY).await;
                        continue;
                    }
                }
                break;
            }
        });
    }
}

fn send_update_via_sender(
    sender: &mpsc::UnboundedSender<EngineMessage>,
    request_id: String,
    status: TransactionStatus,
    tx_hash: Option<String>,
    error: Option<String>,
) {
    let update = TransactionUpdate {
        request_id,
        status,
        tx_hash,
        error,
        timestamp: timestamp_millis(),
        gas_used: None,
        gas_price: None,
    };

    debug!("Sending transaction update: {:?}", update);

    if let Err(send_error) = sender.send(EngineMessage::TransactionUpdate(update)) {
        error!("Failed to send transaction update: {}", send_error);
    }
}

fn is_terminal_relay_error(error: &str) -> bool {
    error.contains("direct submitter transaction reverted")
        || error.contains("aws_kms submitter transaction reverted")
}

fn timestamp_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system clock must not be before Unix epoch")
        .as_millis() as u64
}

#[cfg(test)]
mod tests {
    use super::*;
    use alloy::primitives::{keccak256, Address, TxHash, U256};
    use alloy::sol_types::SolCall;
    use async_trait::async_trait;
    use hypercall_transaction_submitter_db::{
        PendingSubmissionRow, SubmissionDetailRow, TransactionSubmitterReader,
    };
    use hypercall_types::topics::TOPIC_TRANSACTION_UPDATES;
    use hypercall_types::TransactionType;
    use hypercall_types::WalletAddress;
    use std::collections::HashMap;
    use std::str::FromStr;
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
    use tokio::sync::{Mutex, Notify};
    use tokio::time::{timeout, Duration};

    const ACCOUNT: &str = "0x1111111111111111111111111111111111111111";

    enum RelayerBehavior {
        Success,
        Pending,
        Fail,
        Reverted,
    }

    #[derive(Clone)]
    #[allow(dead_code)]
    enum PendingReconciliationBehavior {
        Unsupported,
        StillPending(&'static str),
        Confirmed,
        Failed(&'static str),
    }

    type SharedRequests = Arc<Mutex<Vec<ContractCall>>>;

    struct MockEventBus {
        sender: mpsc::UnboundedSender<EngineMessage>,
        receiver: Mutex<Option<mpsc::UnboundedReceiver<EngineMessage>>>,
        subscribers: Mutex<Vec<(Vec<String>, mpsc::UnboundedSender<EngineMessage>)>>,
    }

    impl MockEventBus {
        fn new() -> Self {
            let (sender, receiver) = mpsc::unbounded_channel();
            Self {
                sender,
                receiver: Mutex::new(Some(receiver)),
                subscribers: Mutex::new(Vec::new()),
            }
        }

        async fn start_processing(self: Arc<Self>) {
            let Some(mut receiver) = self.receiver.lock().await.take() else {
                return;
            };
            tokio::spawn(async move {
                while let Some(message) = receiver.recv().await {
                    let topic = match &message {
                        EngineMessage::TransactionRequest(_) => TOPIC_TRANSACTION_REQUESTS,
                        EngineMessage::TransactionUpdate(_) => TOPIC_TRANSACTION_UPDATES,
                        _ => continue,
                    };
                    let subscribers = self.subscribers.lock().await;
                    for (topics, sender) in subscribers.iter() {
                        if topics.iter().any(|candidate| candidate == topic) {
                            let _ = sender.send(message.clone());
                        }
                    }
                }
            });
        }
    }

    #[async_trait]
    impl EventBusTrait for MockEventBus {
        fn get_sender(&self) -> mpsc::UnboundedSender<EngineMessage> {
            self.sender.clone()
        }

        async fn subscribe(
            &self,
            topics: Vec<String>,
        ) -> Result<mpsc::UnboundedReceiver<EngineMessage>, String> {
            let (sender, receiver) = mpsc::unbounded_channel();
            self.subscribers.lock().await.push((topics, sender));
            Ok(receiver)
        }
    }

    struct CapturingTxRelayer {
        behavior: RelayerBehavior,
        pending_reconciliation: PendingReconciliationBehavior,
        requests: SharedRequests,
    }

    impl CapturingTxRelayer {
        fn with_shared_capture(
            behavior: RelayerBehavior,
            pending_reconciliation: PendingReconciliationBehavior,
            requests: SharedRequests,
        ) -> Self {
            Self {
                behavior,
                pending_reconciliation,
                requests,
            }
        }
    }

    #[async_trait]
    impl TxRelayerTrait for CapturingTxRelayer {
        fn submitter_address(&self) -> SubmitterId {
            Address::repeat_byte(0x42)
        }

        async fn send_transaction(
            &self,
            request: ContractCall,
            db_nonce_floor: Option<u64>,
        ) -> EyreResult<SubmittedTransaction> {
            self.requests.lock().await.push(request);
            let nonce = db_nonce_floor.unwrap_or(7);

            match self.behavior {
                RelayerBehavior::Success => Ok(SubmittedTransaction::new(
                    self.submitter_address(),
                    nonce,
                    TxHash::from_str(
                        "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                    )
                    .expect("valid tx hash")
                    .to_string(),
                    true,
                )),
                RelayerBehavior::Pending => Ok(SubmittedTransaction::new(
                    self.submitter_address(),
                    nonce,
                    TxHash::from_str(
                        "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
                    )
                    .expect("valid tx hash")
                    .to_string(),
                    false,
                )),
                RelayerBehavior::Fail => Err(eyre::eyre!("forced failure")),
                RelayerBehavior::Reverted => Err(eyre::eyre!(
                    "direct submitter transaction reverted: hash=0xdddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
                )),
            }
        }

        async fn reconcile_pending_transaction(
            &self,
            _tx_hash: &str,
        ) -> EyreResult<PendingTransactionStatus> {
            Ok(match self.pending_reconciliation {
                PendingReconciliationBehavior::Unsupported => PendingTransactionStatus::Unsupported,
                PendingReconciliationBehavior::StillPending(reason) => {
                    PendingTransactionStatus::Pending(reason.to_string())
                }
                PendingReconciliationBehavior::Confirmed => {
                    PendingTransactionStatus::Confirmed(_tx_hash.to_string())
                }
                PendingReconciliationBehavior::Failed(error) => {
                    PendingTransactionStatus::Failed(error.to_string())
                }
            })
        }
    }

    async fn take_requests(shared_requests: &SharedRequests) -> Vec<ContractCall> {
        shared_requests.lock().await.clone()
    }

    struct BlockingPendingRelayer {
        started: Arc<Notify>,
        release: Arc<Notify>,
        requests: SharedRequests,
    }

    #[async_trait]
    impl TxRelayerTrait for BlockingPendingRelayer {
        fn submitter_address(&self) -> SubmitterId {
            Address::repeat_byte(0x43)
        }

        async fn send_transaction(
            &self,
            request: ContractCall,
            db_nonce_floor: Option<u64>,
        ) -> EyreResult<SubmittedTransaction> {
            self.requests.lock().await.push(request);
            Ok(SubmittedTransaction::new(
                self.submitter_address(),
                db_nonce_floor.unwrap_or(8),
                TxHash::from_str(
                    "0xcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
                )
                .expect("valid tx hash")
                .to_string(),
                false,
            ))
        }

        async fn reconcile_pending_transaction(
            &self,
            _tx_hash: &str,
        ) -> EyreResult<PendingTransactionStatus> {
            self.started.notify_one();
            self.release.notified().await;
            Ok(PendingTransactionStatus::Confirmed(_tx_hash.to_string()))
        }
    }

    struct ReplacementPendingRelayer {
        requests: SharedRequests,
        expected_attempts: Vec<String>,
        confirmed_hash: String,
    }

    #[async_trait]
    impl TxRelayerTrait for ReplacementPendingRelayer {
        fn submitter_address(&self) -> SubmitterId {
            Address::repeat_byte(0x44)
        }

        async fn send_transaction(
            &self,
            request: ContractCall,
            db_nonce_floor: Option<u64>,
        ) -> EyreResult<SubmittedTransaction> {
            self.requests.lock().await.push(request);
            Ok(SubmittedTransaction::with_hashes(
                self.submitter_address(),
                db_nonce_floor.unwrap_or(9),
                self.expected_attempts
                    .last()
                    .expect("test attempts should not be empty")
                    .clone(),
                self.expected_attempts.clone(),
                false,
            ))
        }

        async fn reconcile_pending_transactions(
            &self,
            tx_hashes: &[String],
        ) -> EyreResult<PendingTransactionStatus> {
            assert_eq!(tx_hashes, self.expected_attempts.as_slice());
            Ok(PendingTransactionStatus::Confirmed(
                self.confirmed_hash.clone(),
            ))
        }
    }

    struct BlockingFirstSendRelayer {
        select_count: AtomicU64,
        send_count: AtomicU64,
        first_send_started: Notify,
        second_select_started: Notify,
        release_first_send: Notify,
    }

    impl BlockingFirstSendRelayer {
        fn new() -> Self {
            Self {
                select_count: AtomicU64::new(0),
                send_count: AtomicU64::new(0),
                first_send_started: Notify::new(),
                second_select_started: Notify::new(),
                release_first_send: Notify::new(),
            }
        }
    }

    struct RecorderOrderingRelayer {
        recorded_before_send: Arc<AtomicBool>,
    }

    #[async_trait]
    impl TxRelayerTrait for RecorderOrderingRelayer {
        fn submitter_address(&self) -> SubmitterId {
            Address::repeat_byte(0x46)
        }

        async fn send_transaction_with_nonce_recording_attempts(
            &self,
            _request: ContractCall,
            nonce: SubmittedNonce,
            recorder: Arc<dyn SignedAttemptRecorder>,
        ) -> EyreResult<SubmittedTransaction> {
            let tx_hash = TxHash::from([0x46; 32]).to_string();
            recorder
                .record_signed_attempt(SignedTransactionAttempt {
                    submitter: self.submitter_address(),
                    nonce,
                    tx_hash: tx_hash.clone(),
                    raw_tx: b"raw-signed-test-tx".to_vec(),
                })
                .await?;
            self.recorded_before_send.store(true, Ordering::SeqCst);
            Ok(SubmittedTransaction::new(
                self.submitter_address(),
                nonce,
                tx_hash,
                false,
            ))
        }
    }

    #[derive(Default)]
    struct InMemorySubmitterStore {
        rows: Mutex<HashMap<(SubmitterId, SubmittedNonce), SubmissionDetailRow>>,
    }

    #[async_trait]
    impl TransactionSubmitterReader for InMemorySubmitterStore {
        async fn get_submission_by_nonce(
            &self,
            submitter: &SubmitterId,
            nonce: SubmittedNonce,
        ) -> anyhow::Result<Option<SubmissionDetailRow>> {
            Ok(self.rows.lock().await.get(&(*submitter, nonce)).cloned())
        }
    }

    #[async_trait]
    impl AsyncTransactionSubmitterStore for InMemorySubmitterStore {
        async fn max_nonce_for_submitter(
            &self,
            submitter: &SubmitterId,
        ) -> anyhow::Result<Option<u64>> {
            Ok(self
                .rows
                .lock()
                .await
                .iter()
                .filter(|((row_submitter, _), row)| {
                    row_submitter == submitter && row.primary_tx_hash.is_some()
                })
                .map(|((_, nonce), _)| *nonce)
                .max())
        }

        async fn record_submission(
            &self,
            record: &SubmissionRecord,
            attempts: &[SubmissionAttemptRecord],
        ) -> anyhow::Result<()> {
            let mut rows = self.rows.lock().await;
            let row = rows
                .entry((record.submitter, record.nonce))
                .or_insert_with(|| SubmissionDetailRow {
                    submitter: record.submitter,
                    nonce: record.nonce,
                    status: record.status,
                    primary_tx_hash: record.primary_tx_hash.clone(),
                    terminal_error: record.terminal_error.clone(),
                    attempts: Vec::new(),
                });
            row.status = record.status;
            if record.primary_tx_hash.is_some() {
                row.primary_tx_hash = record.primary_tx_hash.clone();
            }
            row.terminal_error = record.terminal_error.clone();
            for attempt in attempts {
                if !row
                    .attempts
                    .iter()
                    .any(|existing| existing.tx_hash == attempt.tx_hash)
                {
                    row.attempts.push(attempt.clone());
                }
            }
            Ok(())
        }

        async fn update_submission_status(
            &self,
            submitter: &SubmitterId,
            nonce: SubmittedNonce,
            status: SubmissionStatus,
            primary_tx_hash: Option<&str>,
            terminal_error: Option<&str>,
        ) -> anyhow::Result<()> {
            let mut rows = self.rows.lock().await;
            let row = rows
                .get_mut(&(*submitter, nonce))
                .ok_or_else(|| anyhow::anyhow!("missing submission"))?;
            row.status = status;
            if let Some(primary_tx_hash) = primary_tx_hash {
                row.primary_tx_hash = Some(primary_tx_hash.to_string());
            }
            row.terminal_error = terminal_error.map(str::to_string);
            Ok(())
        }

        async fn list_pending_submissions(
            &self,
            _after_submission_id: i64,
            _limit: i64,
        ) -> anyhow::Result<Vec<PendingSubmissionRow>> {
            Ok(Vec::new())
        }
    }

    #[async_trait]
    impl TxRelayerTrait for BlockingFirstSendRelayer {
        fn submitter_address(&self) -> SubmitterId {
            Address::repeat_byte(0x45)
        }

        async fn select_next_nonce(
            &self,
            db_nonce_floor: Option<u64>,
        ) -> EyreResult<SubmittedNonce> {
            let select_count = self.select_count.fetch_add(1, Ordering::SeqCst) + 1;
            if select_count == 2 {
                self.second_select_started.notify_one();
            }
            Ok(db_nonce_floor.unwrap_or(21))
        }

        async fn send_transaction_with_nonce(
            &self,
            _request: ContractCall,
            nonce: SubmittedNonce,
        ) -> EyreResult<SubmittedTransaction> {
            let send_count = self.send_count.fetch_add(1, Ordering::SeqCst) + 1;
            if send_count == 1 {
                self.first_send_started.notify_one();
                self.release_first_send.notified().await;
                return Err(eyre::eyre!("forced pre-broadcast failure"));
            }

            Ok(SubmittedTransaction::new(
                self.submitter_address(),
                nonce,
                TxHash::from([0x45; 32]).to_string(),
                true,
            ))
        }
    }

    fn rsm_directive_bytes() -> Vec<u8> {
        vec![0x11, 0x22, 0x33]
    }

    fn build_rsm_tx_request(
        request_id: &str,
        signature: String,
        expires_at: u64,
    ) -> TransactionRequest {
        let account = WalletAddress::from_str(ACCOUNT).expect("valid account");
        TransactionRequest {
            request_id: request_id.to_string(),
            wallet_address: account,
            account_contract: account,
            transaction_type: TransactionType::RsmDirective(SignedDirectiveTx {
                directive: rsm_directive_bytes(),
                signature,
            }),
            timestamp: timestamp_millis(),
            expires_at,
        }
    }

    fn contract_call_for_test(external_id: &str) -> ContractCall {
        ContractCall {
            to: test_exchange_address(),
            value: U256::ZERO,
            data: Bytes::from(vec![0x12, 0x34]),
            external_id: Some(external_id.to_string()),
        }
    }

    async fn recv_tx_update(
        rx: &mut tokio::sync::mpsc::UnboundedReceiver<EngineMessage>,
    ) -> TransactionUpdate {
        let message = timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("should receive tx update within timeout")
            .expect("channel should remain open");
        match message {
            EngineMessage::TransactionUpdate(update) => update,
            _ => panic!("expected TransactionUpdate"),
        }
    }

    fn valid_signature_hex() -> String {
        format!("0x{}", hex::encode([0xAB; 65]))
    }

    fn valid_signature_bytes() -> Vec<u8> {
        vec![0xAB; 65]
    }

    fn test_exchange_address() -> Address {
        Address::from_str("0x1d70Ff185F6C25E4d76163F16563D63d5b590Cbc")
            .expect("valid exchange address")
    }

    #[tokio::test]
    async fn transaction_submitter_rsm_expired_emits_expired_only() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Success,
                PendingReconciliationBehavior::Unsupported,
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req =
            build_rsm_tx_request("expired-rsm", valid_signature_hex(), now.saturating_sub(1));
        submitter.process_transaction_request(tx_req).await;

        let first = recv_tx_update(&mut updates_rx).await;
        assert_eq!(first.request_id, "expired-rsm");
        assert_eq!(first.status, TransactionStatus::Expired);
        assert_eq!(
            first.error.as_deref(),
            Some("Transaction request expired"),
            "expired message should be explicit"
        );
        assert!(
            timeout(Duration::from_millis(100), updates_rx.recv())
                .await
                .is_err(),
            "expired path should emit exactly one update"
        );

        let requests = take_requests(&shared_requests).await;
        assert!(
            requests.is_empty(),
            "expired requests must not be sent to relayer"
        );
    }

    #[tokio::test]
    async fn transaction_submitter_rsm_success_emits_confirmed_after_relay() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Success,
                PendingReconciliationBehavior::Unsupported,
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let expected_directive = rsm_directive_bytes();
        let expected_signature_bytes = valid_signature_bytes();
        let tx_req = build_rsm_tx_request("success-rsm", valid_signature_hex(), now + 60_000);
        submitter.process_transaction_request(tx_req).await;

        let confirmed = recv_tx_update(&mut updates_rx).await;
        assert_eq!(confirmed.request_id, "success-rsm");
        assert_eq!(confirmed.status, TransactionStatus::Confirmed);
        assert_eq!(
            confirmed.tx_hash.as_deref(),
            Some("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
        );

        let requests = take_requests(&shared_requests).await;
        assert_eq!(
            requests.len(),
            1,
            "exactly one relay request should be sent"
        );
        assert_eq!(requests[0].external_id.as_deref(), Some("success-rsm"));

        let to: Address = requests[0].to.into();
        assert_eq!(to, test_exchange_address());

        let calldata: Bytes = requests[0].data.clone().into();
        let expected_selector = keccak256("executeRsmDirective(bytes,bytes)".as_bytes());
        assert_eq!(&calldata[..4], &expected_selector[..4]);

        let decoded_call = super::exchange_encoder::Exchange::executeRsmDirectiveCall::abi_decode(
            calldata.as_ref(),
        )
        .expect("calldata should decode as executeRsmDirectiveCall");
        assert_eq!(decoded_call.directive, Bytes::from(expected_directive));
        assert_eq!(
            decoded_call.signature,
            Bytes::from(expected_signature_bytes)
        );
    }

    #[tokio::test]
    async fn transaction_submitter_rsm_relayer_error_leaves_outbox_retryable() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Fail,
                PendingReconciliationBehavior::Unsupported,
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request("failed-rsm", valid_signature_hex(), now + 60_000);
        submitter.process_transaction_request(tx_req).await;

        let retryable = recv_tx_update(&mut updates_rx).await;
        assert_eq!(retryable.request_id, "failed-rsm");
        assert_eq!(retryable.status, TransactionStatus::Submitted);
        assert!(retryable.tx_hash.is_none());
        assert!(
            retryable
                .error
                .as_deref()
                .unwrap_or_default()
                .contains("forced failure"),
            "relay transport error should be recorded on a retryable nonterminal update: {retryable:?}"
        );

        let requests = take_requests(&shared_requests).await;
        assert_eq!(
            requests.len(),
            1,
            "relay request should have been attempted"
        );
    }

    #[tokio::test]
    async fn transaction_submitter_mined_revert_emits_terminal_failed_update() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Reverted,
                PendingReconciliationBehavior::Unsupported,
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request("reverted-rsm", valid_signature_hex(), now + 60_000);
        submitter.process_transaction_request(tx_req).await;

        let failed = recv_tx_update(&mut updates_rx).await;
        assert_eq!(failed.request_id, "reverted-rsm");
        assert_eq!(failed.status, TransactionStatus::Failed);
        assert!(failed.tx_hash.is_none());
        assert!(
            failed
                .error
                .as_deref()
                .unwrap_or_default()
                .contains("direct submitter transaction reverted"),
            "mined reverts must be terminal failures for manual reconciliation: {failed:?}"
        );

        let requests = take_requests(&shared_requests).await;
        assert_eq!(
            requests.len(),
            1,
            "relay request should have been attempted"
        );
    }

    #[tokio::test]
    async fn transaction_submitter_pending_relayer_emits_pending_after_relay() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Pending,
                PendingReconciliationBehavior::Unsupported,
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request("pending-rsm", valid_signature_hex(), now + 60_000);
        submitter.process_transaction_request(tx_req).await;

        let pending = recv_tx_update(&mut updates_rx).await;
        assert_eq!(pending.request_id, "pending-rsm");
        assert_eq!(pending.status, TransactionStatus::Pending);
        assert_eq!(
            pending.tx_hash.as_deref(),
            Some("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
        );
    }

    #[tokio::test]
    async fn transaction_submitter_pending_relayer_with_reconciliation_timeout_stays_pending() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Pending,
                PendingReconciliationBehavior::StillPending("timed out waiting for receipt"),
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request(
            "pending-reconciled-rsm",
            valid_signature_hex(),
            now + 60_000,
        );
        submitter.process_transaction_request(tx_req).await;

        let pending = recv_tx_update(&mut updates_rx).await;
        assert_eq!(pending.request_id, "pending-reconciled-rsm");
        assert_eq!(pending.status, TransactionStatus::Pending);

        let still_pending = recv_tx_update(&mut updates_rx).await;
        assert_eq!(still_pending.request_id, "pending-reconciled-rsm");
        assert_eq!(still_pending.status, TransactionStatus::Pending);
        assert_eq!(
            still_pending.error.as_deref(),
            Some("timed out waiting for receipt")
        );
        assert_eq!(
            still_pending.tx_hash.as_deref(),
            Some("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
        );
    }

    #[tokio::test]
    async fn transaction_submitter_pending_reconciliation_does_not_block_hot_path() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let started = Arc::new(Notify::new());
        let release = Arc::new(Notify::new());
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(BlockingPendingRelayer {
                started: Arc::clone(&started),
                release: Arc::clone(&release),
                requests: shared_requests.clone(),
            }),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request("pending-hot-path", valid_signature_hex(), now + 60_000);
        let handle = tokio::spawn(async move {
            submitter.process_transaction_request(tx_req).await;
        });

        let pending = recv_tx_update(&mut updates_rx).await;
        assert_eq!(pending.request_id, "pending-hot-path");
        assert_eq!(pending.status, TransactionStatus::Pending);

        timeout(Duration::from_millis(200), started.notified())
            .await
            .expect("reconciliation task should start in background");
        timeout(Duration::from_millis(200), handle)
            .await
            .expect("hot path should return before reconciliation completes")
            .expect("transaction task should complete successfully");

        assert!(
            timeout(Duration::from_millis(100), updates_rx.recv())
                .await
                .is_err(),
            "confirmed update should wait for reconciliation completion"
        );

        release.notify_one();

        let confirmed = recv_tx_update(&mut updates_rx).await;
        assert_eq!(confirmed.request_id, "pending-hot-path");
        assert_eq!(confirmed.status, TransactionStatus::Confirmed);
        assert_eq!(
            confirmed.tx_hash.as_deref(),
            Some("0xcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc")
        );
    }

    #[tokio::test]
    async fn transaction_submitter_serializes_nonce_selection_through_first_submit() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let relayer = Arc::new(BlockingFirstSendRelayer::new());
        let submitter = Arc::new(TransactionSubmitter {
            event_bus,
            tx_relayer: relayer.clone(),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        });

        let first_submitter = submitter.clone();
        let first = tokio::spawn(async move {
            first_submitter
                .submit_contract_call("first", contract_call_for_test("first"))
                .await
        });

        timeout(
            Duration::from_secs(1),
            relayer.first_send_started.notified(),
        )
        .await
        .expect("first submit should enter send path");

        let second_submitter = submitter.clone();
        let second = tokio::spawn(async move {
            second_submitter
                .submit_contract_call("second", contract_call_for_test("second"))
                .await
        });

        assert!(
            timeout(
                Duration::from_millis(100),
                relayer.second_select_started.notified()
            )
            .await
            .is_err(),
            "second submit must not allocate a nonce while first submit is undecided"
        );

        relayer.release_first_send.notify_one();

        let first_result = first.await.expect("first task should join");
        assert!(first_result.is_err());

        let second_result = second.await.expect("second task should join");
        assert_eq!(
            second_result.expect("second submit should succeed").nonce,
            21
        );
        assert_eq!(relayer.select_count.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn transaction_submitter_persists_signed_attempt_before_send() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let store = Arc::new(InMemorySubmitterStore::default());
        let recorded_before_send = Arc::new(AtomicBool::new(false));
        let submitter = TransactionSubmitter {
            event_bus,
            tx_relayer: Arc::new(RecorderOrderingRelayer {
                recorded_before_send: recorded_before_send.clone(),
            }),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: Some(store.clone()),
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let submission = submitter
            .submit_contract_call(
                "record-before-send",
                contract_call_for_test("record-before-send"),
            )
            .await
            .expect("submission should stay pending");

        assert!(
            recorded_before_send.load(Ordering::SeqCst),
            "relayer must not send until the recorder accepted the signed attempt"
        );
        let detail = store
            .get_submission_by_nonce(&Address::repeat_byte(0x46), submission.nonce)
            .await
            .expect("db read")
            .expect("submission exists");
        assert_eq!(detail.status, SubmissionStatus::Broadcasted);
        assert_eq!(
            detail.primary_tx_hash.as_deref(),
            Some(submission.hash.as_str())
        );
        assert_eq!(detail.attempts.len(), 1);
        assert_eq!(detail.attempts[0].raw_tx, b"raw-signed-test-tx");
    }

    #[tokio::test]
    async fn transaction_submitter_reconciles_all_replacement_attempt_hashes() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let attempts = vec![
            "0x1111111111111111111111111111111111111111111111111111111111111111".to_string(),
            "0x2222222222222222222222222222222222222222222222222222222222222222".to_string(),
        ];
        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(ReplacementPendingRelayer {
                requests: shared_requests.clone(),
                expected_attempts: attempts.clone(),
                confirmed_hash: attempts[0].clone(),
            }),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request(
            "replacement-attempts-rsm",
            valid_signature_hex(),
            now + 60_000,
        );
        submitter.process_transaction_request(tx_req).await;

        let pending = recv_tx_update(&mut updates_rx).await;
        assert_eq!(pending.request_id, "replacement-attempts-rsm");
        assert_eq!(pending.status, TransactionStatus::Pending);
        assert_eq!(
            pending.tx_hash.as_deref(),
            Some(attempts[1].as_str()),
            "hot path should expose the newest replacement hash while reconciliation runs"
        );

        let confirmed = recv_tx_update(&mut updates_rx).await;
        assert_eq!(confirmed.request_id, "replacement-attempts-rsm");
        assert_eq!(confirmed.status, TransactionStatus::Confirmed);
        assert_eq!(
            confirmed.tx_hash.as_deref(),
            Some(attempts[0].as_str()),
            "reconciliation must finalize with the attempt that actually mined"
        );

        let requests = take_requests(&shared_requests).await;
        assert_eq!(requests.len(), 1);
    }

    #[tokio::test]
    async fn transaction_submitter_rsm_invalid_signature_emits_terminal_failure() {
        let event_bus = Arc::new(MockEventBus::new());
        event_bus.clone().start_processing().await;
        let mut updates_rx = event_bus
            .subscribe(vec![TOPIC_TRANSACTION_UPDATES.to_string()])
            .await
            .expect("subscribe should succeed");

        let shared_requests: SharedRequests = Arc::new(Mutex::new(Vec::new()));
        let submitter = TransactionSubmitter {
            event_bus: event_bus.clone(),
            tx_relayer: Arc::new(CapturingTxRelayer::with_shared_capture(
                RelayerBehavior::Success,
                PendingReconciliationBehavior::Unsupported,
                shared_requests.clone(),
            )),
            exchange_address: test_exchange_address(),
            db: None,
            submitter_store: None,
            nonce_lock: Arc::new(Mutex::new(())),
        };

        let now = timestamp_millis();
        let tx_req = build_rsm_tx_request("invalid-sig-rsm", "0xzz".to_string(), now + 60_000);
        submitter.process_transaction_request(tx_req).await;

        let failed = recv_tx_update(&mut updates_rx).await;
        assert_eq!(failed.request_id, "invalid-sig-rsm");
        assert_eq!(failed.status, TransactionStatus::Failed);
        assert!(
            failed
                .error
                .as_deref()
                .is_some_and(|error| error.contains("invalid signature bytes")),
            "signature parse errors must emit an explicit terminal failure"
        );

        let requests = take_requests(&shared_requests).await;
        assert!(
            requests.is_empty(),
            "invalid signatures must not be sent to relayer"
        );
    }
}