diff --git a/crates/cli/src/commands/tx.rs b/crates/cli/src/commands/tx.rs index c55e28c7..9cdfe8e2 100644 --- a/crates/cli/src/commands/tx.rs +++ b/crates/cli/src/commands/tx.rs @@ -14,7 +14,8 @@ use rrelayer_core::{ common_types::EvmAddress, relayer::RelayerId, transaction::types::{ - Transaction, TransactionData, TransactionId, TransactionSpeed, TransactionValue, + Transaction, TransactionData, TransactionHash, TransactionId, TransactionSpeed, + TransactionValue, }, }; use std::io::{self, Write}; @@ -218,7 +219,7 @@ async fn handle_withdraw( println!("ERC-20 token withdrawal transaction sent.."); println!("Transaction id: {}", tx.id); - println!("Transaction hash: {}", tx.hash); + println!("Transaction hash: {}", format_transaction_hash(tx.hash)); } None => { let tx = client @@ -238,7 +239,7 @@ async fn handle_withdraw( println!("ETH withdrawal transaction sent.."); println!("Transaction id: {}", tx.id); - println!("Transaction hash: {}", tx.hash); + println!("Transaction hash: {}", format_transaction_hash(tx.hash)); } } @@ -369,11 +370,15 @@ async fn handle_send( println!("Transaction sent.."); println!("Transaction id: {}", tx.id); - println!("Transaction hash: {}", tx.hash); + println!("Transaction hash: {}", format_transaction_hash(tx.hash)); Ok(()) } +fn format_transaction_hash(hash: Option) -> String { + hash.map_or_else(|| "".to_string(), |hash| hash.to_string()) +} + async fn handle_fund( relayer_id: &RelayerId, amount: &str, @@ -619,3 +624,13 @@ fn log_transactions(transactions: Vec) -> Result<(), TransactionErr fn format_time(time: &DateTime) -> String { time.format("%Y-%m-%d %H:%M:%S UTC").to_string() } + +#[cfg(test)] +mod tests { + use super::format_transaction_hash; + + #[test] + fn accepted_submission_without_hash_is_printed_as_pending() { + assert_eq!(format_transaction_hash(None), ""); + } +} diff --git a/crates/core/src/provider/evm_provider.rs b/crates/core/src/provider/evm_provider.rs index 8d4e70de..ac40321a 100644 --- a/crates/core/src/provider/evm_provider.rs +++ b/crates/core/src/provider/evm_provider.rs @@ -31,7 +31,7 @@ use alloy::{ eips::{BlockId, BlockNumberOrTag}, network::Ethereum, network::TransactionBuilderError, - primitives::Signature, + primitives::{Bytes, Signature}, providers::{Provider, ProviderBuilder}, rpc::types::TransactionRequest, signers::local::LocalSignerError, @@ -54,6 +54,27 @@ pub type RelayerProvider = Box + Send + Sync>; const BLOCK_GAS_LIMIT_CACHE_TTL: Duration = Duration::from_secs(600); +/// The exact EIP-2718 payload prepared for broadcast and its deterministic hash. +/// +/// Keeping the bytes and hash together prevents recovery evidence from drifting +/// from the payload handed to the RPC provider. +#[derive(Clone, Debug)] +pub struct SignedTransaction { + bytes: Bytes, + hash: TransactionHash, +} + +impl SignedTransaction { + pub fn hash(&self) -> TransactionHash { + self.hash + } + + #[cfg(test)] + pub(crate) fn for_test(hash: TransactionHash) -> Self { + Self { bytes: Bytes::new(), hash } + } +} + #[derive(Clone)] struct BlockGasLimitCache { gas_limit: GasLimit, @@ -349,6 +370,16 @@ impl EvmProvider { Ok(receipt) } + /// Returns true when any configured RPC knows the transaction, false only + /// when every configured RPC conclusively reports it absent, and an error + /// when absence cannot be proven. + pub async fn transaction_exists( + &self, + transaction_hash: &TransactionHash, + ) -> Result> { + transaction_exists_across_clients(&self.rpc_clients, transaction_hash).await + } + pub async fn get_nonce( &self, relayer: &Relayer, @@ -361,10 +392,7 @@ impl EvmProvider { WalletOrProviderError::InternalError(format!("Failed to get address: {}", e)) })?; - let nonce = self - .rpc_client() - .get_transaction_count(address.into_address()) - .block_id(BlockId::Number(BlockNumberOrTag::Pending)) + let nonce = pending_nonce_across_clients(&self.rpc_clients, &address) .await .map_err(WalletOrProviderError::ProviderError)?; @@ -375,11 +403,7 @@ impl EvmProvider { &self, address: &EvmAddress, ) -> Result> { - let nonce = self - .rpc_client() - .get_transaction_count(address.into_address()) - .block_id(BlockId::Number(BlockNumberOrTag::Pending)) - .await?; + let nonce = pending_nonce_across_clients(&self.rpc_clients, address).await?; Ok(TransactionNonce::new(nonce)) } @@ -389,12 +413,21 @@ impl EvmProvider { relayer: &Relayer, transaction: TypedTransaction, ) -> Result { + let signed = self.prepare_signed_transaction(relayer, &transaction).await?; + self.send_raw_transaction(&signed).await + } + + pub async fn prepare_signed_transaction( + &self, + relayer: &Relayer, + transaction: &TypedTransaction, + ) -> Result { let signature = self - .sign_transaction(relayer, &transaction) + .sign_transaction(relayer, transaction) .await .map_err(|e| SendTransactionError::InternalError(e.to_string()))?; - self.send_signed_transaction(transaction, signature).await + Ok(Self::signed_transaction(transaction.clone(), signature)) } pub async fn send_signed_transaction( @@ -402,6 +435,14 @@ impl EvmProvider { transaction: TypedTransaction, signature: Signature, ) -> Result { + let signed = Self::signed_transaction(transaction, signature); + self.send_raw_transaction(&signed).await + } + + fn signed_transaction( + transaction: TypedTransaction, + signature: Signature, + ) -> SignedTransaction { let tx_envelope = match transaction { TypedTransaction::Legacy(tx) => TxEnvelope::Legacy(tx.into_signed(signature)), TypedTransaction::Eip2930(tx) => TxEnvelope::Eip2930(tx.into_signed(signature)), @@ -410,12 +451,22 @@ impl EvmProvider { TypedTransaction::Eip7702(tx) => TxEnvelope::Eip7702(tx.into_signed(signature)), }; - let provider = self.rpc_client(); - let tx_bytes = tx_envelope.encoded_2718(); + // For EIP-4844, the bytes sent over the wire include the blob sidecar, + // but the transaction identity is the hash of the signed inner + // transaction. Alloy caches that consensus hash on the envelope. + let hash = TransactionHash::from_alloy_hash(tx_envelope.hash()); + let bytes = Bytes::from(tx_envelope.encoded_2718()); + + SignedTransaction { bytes, hash } + } - let receipt = provider.send_raw_transaction(&tx_bytes).await?; + pub async fn send_raw_transaction( + &self, + transaction: &SignedTransaction, + ) -> Result { + let _ = self.rpc_client().send_raw_transaction(&transaction.bytes).await?; - Ok(TransactionHash::from_alloy_hash(receipt.tx_hash())) + Ok(transaction.hash) } pub async fn sign_transaction( @@ -569,17 +620,220 @@ impl EvmProvider { } } +async fn pending_nonce_across_clients( + rpc_clients: &[Arc], + address: &EvmAddress, +) -> Result> { + if rpc_clients.is_empty() { + return Err(RpcError::Transport(TransportErrorKind::Custom( + "no RPC providers configured".to_string().into(), + ))); + } + + let mut max_nonce: Option = None; + let mut first_error = None; + + for rpc_client in rpc_clients { + match rpc_client + .get_transaction_count(address.into_address()) + .block_id(BlockId::Number(BlockNumberOrTag::Pending)) + .await + { + Ok(nonce) => max_nonce = Some(max_nonce.map_or(nonce, |max| max.max(nonce))), + Err(error) if first_error.is_none() => first_error = Some(error), + Err(_) => {} + } + } + + if let Some(error) = first_error { + return Err(error); + } + + max_nonce.ok_or_else(|| { + RpcError::Transport(TransportErrorKind::Custom( + "no RPC providers configured".to_string().into(), + )) + }) +} + +async fn transaction_exists_across_clients( + rpc_clients: &[Arc], + transaction_hash: &TransactionHash, +) -> Result> { + if rpc_clients.is_empty() { + return Err(RpcError::Transport(TransportErrorKind::Custom( + "no RPC providers configured".to_string().into(), + ))); + } + + let mut first_error = None; + + for rpc_client in rpc_clients { + match rpc_client.get_transaction_by_hash(transaction_hash.into_alloy_hash()).await { + Ok(Some(_)) => return Ok(true), + Ok(None) => {} + Err(error) if first_error.is_none() => first_error = Some(error), + Err(_) => {} + } + } + + match first_error { + Some(error) => Err(error), + None => Ok(false), + } +} + #[cfg(test)] mod tests { use super::*; use crate::relayer::RelayerId; use crate::wallet::WalletManagerChainId; + use alloy::{ + consensus::{TxEip4844, TxEip4844Variant, TxEip4844WithSidecar}, + primitives::{keccak256, Address, TxHash, U256}, + providers::ProviderBuilder, + transports::mock::Asserter, + }; + use alloy_eips::{eip4844::BlobTransactionSidecar, eip7594::BlobTransactionSidecarVariant}; use async_trait::async_trait; use chrono::Utc; + use serde_json::json; use tokio::sync::Mutex; + fn mock_client(asserter: Asserter) -> Arc { + let provider = + ProviderBuilder::new().network::().connect_mocked_client(asserter); + Arc::new(Box::new(provider)) + } + + fn rpc_transaction(hash: TxHash) -> serde_json::Value { + json!({ + "blockHash": null, + "blockNumber": null, + "hash": hash.to_string(), + "transactionIndex": null, + "type": "0x0", + "nonce": "0x0", + "input": "0x", + "r": "0x3b08715b4403c792b8c7567edea634088bedcd7f60d9352b1f16c69830f3afd5", + "s": "0x10b9afb67d2ec8b956f0e1dbc07eb79152904f3a7bf789fc869db56320adfe09", + "chainId": "0x1", + "v": "0x1c", + "gas": "0x5208", + "from": "0x32be343b94f860124dc4fee278fdcbd38c102d88", + "to": "0xdf190dc7190dfba737d7777a163445b7fff16133", + "value": "0x0", + "gasPrice": "0x1" + }) + } + + #[test] + fn blob_transaction_hash_excludes_the_network_sidecar() { + let transaction = TypedTransaction::Eip4844(TxEip4844Variant::TxEip4844WithSidecar( + TxEip4844WithSidecar { + tx: TxEip4844 { + chain_id: 1, + nonce: 1, + max_priority_fee_per_gas: 1, + max_fee_per_gas: 2, + gas_limit: 100_000, + to: Address::ZERO, + value: U256::ZERO, + access_list: Default::default(), + blob_versioned_hashes: vec![TxHash::repeat_byte(1)], + max_fee_per_blob_gas: 1, + input: Bytes::new(), + }, + sidecar: BlobTransactionSidecarVariant::Eip4844(BlobTransactionSidecar { + blobs: vec![[2; 131_072].into()], + commitments: vec![[3; 48].into()], + proofs: vec![[4; 48].into()], + }), + }, + )); + let signature = Signature::test_signature().with_parity(true); + let expected_envelope = TxEnvelope::Eip4844(match transaction.clone() { + TypedTransaction::Eip4844(tx) => tx.into_signed(signature), + _ => unreachable!(), + }); + let network_payload_hash = keccak256(expected_envelope.encoded_2718()); + + let signed = EvmProvider::signed_transaction(transaction, signature); + + assert_eq!(signed.hash(), TransactionHash::from_alloy_hash(expected_envelope.hash())); + assert_ne!(signed.hash(), TransactionHash::from_alloy_hash(&network_payload_hash)); + } + + #[tokio::test] + async fn transaction_absence_requires_every_configured_provider_to_miss() { + let hash = TxHash::repeat_byte(1); + let first = Asserter::new(); + first.push_success(&serde_json::Value::Null); + let second = Asserter::new(); + second.push_success(&serde_json::Value::Null); + let clients = vec![mock_client(first), mock_client(second)]; + + let exists = transaction_exists_across_clients(&clients, &TransactionHash::new(hash)).await; + + assert!(!exists.unwrap()); + } + + #[tokio::test] + async fn transaction_exists_when_any_configured_provider_finds_hash() { + let hash = TxHash::repeat_byte(2); + let first = Asserter::new(); + first.push_success(&serde_json::Value::Null); + let second = Asserter::new(); + second.push_success(&rpc_transaction(hash)); + let clients = vec![mock_client(first), mock_client(second)]; + + let exists = transaction_exists_across_clients(&clients, &TransactionHash::new(hash)).await; + + assert!(exists.unwrap()); + } + + #[tokio::test] + async fn transaction_absence_fails_closed_when_any_provider_errors() { + let hash = TxHash::repeat_byte(3); + let first = Asserter::new(); + first.push_success(&serde_json::Value::Null); + let second = Asserter::new(); + second.push_failure_msg("backend unavailable"); + let clients = vec![mock_client(first), mock_client(second)]; + + let exists = transaction_exists_across_clients(&clients, &TransactionHash::new(hash)).await; + + assert!(exists.is_err()); + } + + #[tokio::test] + async fn pending_nonce_uses_maximum_only_when_every_provider_responds() { + let first = Asserter::new(); + first.push_success(&"0x7"); + let second = Asserter::new(); + second.push_success(&"0x35"); + let clients = vec![mock_client(first), mock_client(second)]; + + let nonce = pending_nonce_across_clients(&clients, &EvmAddress::zero()).await; + + assert_eq!(nonce.unwrap(), 53); + } + + #[tokio::test] + async fn pending_nonce_fails_closed_when_any_provider_errors() { + let first = Asserter::new(); + first.push_success(&"0x35"); + let second = Asserter::new(); + second.push_failure_msg("backend unavailable"); + let clients = vec![mock_client(first), mock_client(second)]; + + let nonce = pending_nonce_across_clients(&clients, &EvmAddress::zero()).await; + + assert!(nonce.is_err()); + } + struct RecordingWalletManager { - last_create_chain: Arc>>, + last_create_chain: Arc>>, address: EvmAddress, } @@ -587,17 +841,18 @@ mod tests { impl WalletManagerTrait for RecordingWalletManager { async fn create_wallet( &self, - _wallet_index: u32, + wallet_index: u32, chain_id: WalletManagerChainId, ) -> Result { match chain_id { WalletManagerChainId::Cloned(chain) => { let mut last_create_chain = self.last_create_chain.lock().await; - *last_create_chain = Some((chain.cloned_from.u64(), chain.cloned_to.u64())); + *last_create_chain = + Some((wallet_index, chain.cloned_from.u64(), chain.cloned_to.u64())); } WalletManagerChainId::ChainId(chain_id) => { let mut last_create_chain = self.last_create_chain.lock().await; - *last_create_chain = Some((chain_id.u64(), chain_id.u64())); + *last_create_chain = Some((wallet_index, chain_id.u64(), chain_id.u64())); } } @@ -699,6 +954,6 @@ mod tests { let cloned_address = provider.clone_wallet(&source_relayer).await.unwrap(); assert_eq!(cloned_address, address); - assert_eq!(*last_create_chain.lock().await, Some((1, 31337))); + assert_eq!(*last_create_chain.lock().await, Some((7, 1, 31337))); } } diff --git a/crates/core/src/provider/mod.rs b/crates/core/src/provider/mod.rs index 3560b222..96580a45 100644 --- a/crates/core/src/provider/mod.rs +++ b/crates/core/src/provider/mod.rs @@ -12,6 +12,7 @@ use crate::gas::GasEstimatorError; use crate::wallet::get_mnemonic_from_signing_key; pub use evm_provider::{ create_retry_client, EvmProvider, RelayerProvider, RetryClientError, SendTransactionError, + SignedTransaction, }; #[derive(Error, Debug)] diff --git a/crates/core/src/relayer/db/mod.rs b/crates/core/src/relayer/db/mod.rs index d7645260..108ceb66 100644 --- a/crates/core/src/relayer/db/mod.rs +++ b/crates/core/src/relayer/db/mod.rs @@ -1,4 +1,5 @@ mod builders; mod read; +mod wallet_index_allocation; mod write; pub use write::{CreateRelayerError, CreateRelayerMode}; diff --git a/crates/core/src/relayer/db/wallet_index_allocation.rs b/crates/core/src/relayer/db/wallet_index_allocation.rs new file mode 100644 index 00000000..b6002b20 --- /dev/null +++ b/crates/core/src/relayer/db/wallet_index_allocation.rs @@ -0,0 +1,48 @@ +pub(super) fn next_normal_wallet_index_sql(chain_id_parameter: &str) -> String { + format!( + " + SELECT COALESCE(MAX(wallet_index), -1) + 1 AS wallet_index + FROM relayer.record + WHERE chain_id = {chain_id_parameter} + AND is_private_key = FALSE + AND deleted = FALSE + AND wallet_index >= 0 + " + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn next_normal_wallet_index_from_rows(rows: &[(i32, bool, bool)]) -> i32 { + rows.iter() + .filter(|(wallet_index, is_private_key, deleted)| { + *wallet_index >= 0 && !*is_private_key && !*deleted + }) + .map(|(wallet_index, _, _)| *wallet_index) + .max() + .unwrap_or(-1) + + 1 + } + + #[test] + fn private_key_rows_do_not_consume_normal_wallet_indexes() { + assert_eq!(next_normal_wallet_index_from_rows(&[(-1, true, false), (-2, true, false)]), 0); + } + + #[test] + fn deleted_rows_may_be_safely_reused_without_colliding_with_live_rows() { + let rows = [(-1, true, false), (4, false, true), (2, false, false)]; + assert_eq!(next_normal_wallet_index_from_rows(&rows), 3); + } + + #[test] + fn allocation_sql_filters_to_live_normal_namespace() { + let sql = next_normal_wallet_index_sql("$3"); + assert!(sql.contains("chain_id = $3")); + assert!(sql.contains("is_private_key = FALSE")); + assert!(sql.contains("deleted = FALSE")); + assert!(sql.contains("wallet_index >= 0")); + } +} diff --git a/crates/core/src/relayer/db/write.rs b/crates/core/src/relayer/db/write.rs index 7cfc3441..8bcca895 100644 --- a/crates/core/src/relayer/db/write.rs +++ b/crates/core/src/relayer/db/write.rs @@ -11,6 +11,8 @@ use std::error::Error; use thiserror::Error; use tracing::log::error; +use super::wallet_index_allocation::next_normal_wallet_index_sql; + #[derive(Error, Debug)] pub enum CreateRelayerError { #[error("Relayer could not be saved in DB - name: {0}, chainId: {1}: {0}")] @@ -156,21 +158,24 @@ impl PostgresClient { let new_relayer_id_val = new_relayer_id; let name_val = name.to_string(); let chain_id_val = *chain_id; + let next_wallet_index_sql = next_normal_wallet_index_sql("$3"); self.with_transaction(move |tx| { Box::pin(async move { - let query = " + tx.execute("SELECT pg_advisory_xact_lock($1)", &[&chain_id_val]) + .await + .map_err(PostgresError::PgError)?; + + let query = format!(" WITH new_wallet_index AS ( - SELECT COALESCE(MAX(wallet_index), -1) + 1 AS wallet_index - FROM relayer.record - WHERE chain_id = $3 + {next_wallet_index_sql} ) INSERT INTO relayer.record (id, name, chain_id, wallet_index, is_private_key) SELECT $1, $2, $3, wallet_index, false FROM new_wallet_index - RETURNING wallet_index"; + RETURNING wallet_index"); - let rows = tx.query(query, &[&new_relayer_id_val, &name_val, &chain_id_val]).await.map_err(PostgresError::PgError)?; + let rows = tx.query(&query, &[&new_relayer_id_val, &name_val, &chain_id_val]).await.map_err(PostgresError::PgError)?; let wallet_index: i32 = rows.first() .map(|row| row.get("wallet_index")) diff --git a/crates/core/src/relayer/types/relayer.rs b/crates/core/src/relayer/types/relayer.rs index 6dacd357..3ed6993c 100644 --- a/crates/core/src/relayer/types/relayer.rs +++ b/crates/core/src/relayer/types/relayer.rs @@ -50,11 +50,8 @@ pub struct Relayer { impl Relayer { /// Get the WalletIndex enum for this relayer pub fn wallet_index_type(&self) -> WalletIndex { - if self.is_private_key { - WalletIndex::PrivateKey(self.wallet_index) - } else { - WalletIndex::Normal(self.wallet_index as u32) - } + WalletIndex::from_db_value(self.wallet_index, self.is_private_key) + .expect("persisted relayer wallet namespace must be valid") } /// Get the wallet index diff --git a/crates/core/src/relayer/types/wallet_index.rs b/crates/core/src/relayer/types/wallet_index.rs index 7ce70f0e..34fc955c 100644 --- a/crates/core/src/relayer/types/wallet_index.rs +++ b/crates/core/src/relayer/types/wallet_index.rs @@ -1,34 +1,88 @@ use serde::{Deserialize, Serialize}; +use thiserror::Error; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub const MAX_NORMAL_WALLET_INDEX: u32 = i32::MAX as u32; +pub const MIN_PRIVATE_KEY_WALLET_MANAGER_INDEX: u32 = MAX_NORMAL_WALLET_INDEX + 2; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] +pub enum WalletIndexError { + #[error("normal wallet index {index} exceeds maximum database index {max}")] + NormalIndexOutOfRange { index: u32, max: u32 }, + + #[error("normal wallet database index must be non-negative, got {index}")] + NormalDbIndexMustBeNonNegative { index: i32 }, + + #[error("private key wallet database index must be negative, got {index}")] + PrivateKeyIndexMustBeNegative { index: i32 }, + + #[error("private key wallet database index {index} cannot be converted")] + PrivateKeyIndexOutOfRange { index: i32 }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] pub enum WalletIndex { - /// Signing providers wallet with positive index Normal(u32), - /// Private key wallet with negative database index (to avoid a big refactor) PrivateKey(i32), } impl WalletIndex { - /// Get the underlying index value for wallet manager operations + pub fn normal(index: u32) -> Result { + if index > MAX_NORMAL_WALLET_INDEX { + return Err(WalletIndexError::NormalIndexOutOfRange { + index, + max: MAX_NORMAL_WALLET_INDEX, + }); + } + Ok(Self::Normal(index)) + } + + pub fn private_key(db_index: i32) -> Result { + if db_index >= 0 { + return Err(WalletIndexError::PrivateKeyIndexMustBeNegative { index: db_index }); + } + if db_index == i32::MIN { + return Err(WalletIndexError::PrivateKeyIndexOutOfRange { index: db_index }); + } + Ok(Self::PrivateKey(db_index)) + } + + pub fn from_db_value(db_value: i32, is_private_key: bool) -> Result { + if is_private_key { + Self::private_key(db_value) + } else { + let index = u32::try_from(db_value).map_err(|_| { + WalletIndexError::NormalDbIndexMustBeNonNegative { index: db_value } + })?; + Self::normal(index) + } + } + + pub fn is_private_key_manager_index(wallet_index: u32) -> bool { + wallet_index >= MIN_PRIVATE_KEY_WALLET_MANAGER_INDEX + } + pub fn index(&self) -> u32 { match self { WalletIndex::Normal(index) => *index, WalletIndex::PrivateKey(db_index) => { - // Convert negative database index to positive array index using maximum u32 range - // to completely avoid conflicts with mnemonic-derived wallets - // u32::MAX = 4,294,967,295, so we use high range for private keys - // -1 -> 4,294,967,294, -2 -> 4,294,967,293, -3 -> 4,294,967,292, etc. - u32::MAX - (-db_index - 1) as u32 + debug_assert!(*db_index < 0 && *db_index != i32::MIN); + let offset = (-i64::from(*db_index) - 1) as u32; + u32::MAX - offset } } } - /// Check if this is a private key wallet + pub fn private_key_internal_index(&self) -> Option { + match self { + WalletIndex::Normal(_) => None, + WalletIndex::PrivateKey(_) => Some(u32::MAX - self.index()), + } + } + pub fn is_private_key(&self) -> bool { matches!(self, WalletIndex::PrivateKey(_)) } - /// Get the database storage value (i32) pub fn db_value(&self) -> i32 { match self { WalletIndex::Normal(index) => *index as i32, @@ -36,3 +90,65 @@ impl WalletIndex { } } } + +impl<'de> Deserialize<'de> for WalletIndex { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + enum UncheckedWalletIndex { + Normal(u32), + PrivateKey(i32), + } + + match UncheckedWalletIndex::deserialize(deserializer)? { + UncheckedWalletIndex::Normal(index) => { + Self::normal(index).map_err(serde::de::Error::custom) + } + UncheckedWalletIndex::PrivateKey(index) => { + Self::private_key(index).map_err(serde::de::Error::custom) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn live_normal_indexes_use_non_negative_database_namespace() { + let zero = WalletIndex::normal(0).unwrap(); + let max = WalletIndex::normal(MAX_NORMAL_WALLET_INDEX).unwrap(); + assert_eq!((zero.index(), zero.db_value()), (0, 0)); + assert_eq!((max.index(), max.db_value()), (MAX_NORMAL_WALLET_INDEX, i32::MAX)); + assert!(WalletIndex::normal(MAX_NORMAL_WALLET_INDEX + 1).is_err()); + } + + #[test] + fn private_key_rows_remain_in_separate_negative_namespace() { + let first = WalletIndex::private_key(-1).unwrap(); + let second = WalletIndex::private_key(-2).unwrap(); + assert_eq!((first.index(), first.private_key_internal_index()), (u32::MAX, Some(0))); + assert_eq!((second.index(), second.private_key_internal_index()), (u32::MAX - 1, Some(1))); + assert_eq!((first.db_value(), second.db_value()), (-1, -2)); + } + + #[test] + fn persisted_namespace_combinations_are_validated() { + assert!(WalletIndex::from_db_value(7, false).is_ok()); + assert!(WalletIndex::from_db_value(-1, true).is_ok()); + assert!(WalletIndex::from_db_value(-1, false).is_err()); + assert!(WalletIndex::from_db_value(0, true).is_err()); + assert!(WalletIndex::private_key(i32::MIN).is_err()); + } + + #[test] + fn private_key_manager_boundary_cannot_overlap_normal_namespace() { + assert!(!WalletIndex::is_private_key_manager_index(MAX_NORMAL_WALLET_INDEX)); + assert!(!WalletIndex::is_private_key_manager_index(MAX_NORMAL_WALLET_INDEX + 1)); + assert!(WalletIndex::is_private_key_manager_index(MIN_PRIVATE_KEY_WALLET_MANAGER_INDEX)); + assert!(WalletIndex::is_private_key_manager_index(u32::MAX)); + } +} diff --git a/crates/core/src/schema/mod.rs b/crates/core/src/schema/mod.rs index 1eac0f6e..c4fe9e0c 100644 --- a/crates/core/src/schema/mod.rs +++ b/crates/core/src/schema/mod.rs @@ -1,6 +1,7 @@ use crate::schema::v1_0_1::apply_v1_0_1_schema; use crate::schema::v1_0_2::apply_v1_0_2_schema; use crate::schema::v1_0_3::apply_v1_0_3_schema; +use crate::schema::v1_0_4::apply_v1_0_4_schema; use crate::{ postgres::{PostgresClient, PostgresError}, schema::v1_0_0::apply_v1_0_0_schema, @@ -10,6 +11,7 @@ mod v1_0_0; mod v1_0_1; mod v1_0_2; mod v1_0_3; +mod v1_0_4; /// Applies the database schema to the database. pub async fn apply_schema(client: &PostgresClient) -> Result<(), PostgresError> { @@ -17,6 +19,7 @@ pub async fn apply_schema(client: &PostgresClient) -> Result<(), PostgresError> apply_v1_0_1_schema(client).await?; apply_v1_0_2_schema(client).await?; apply_v1_0_3_schema(client).await?; + apply_v1_0_4_schema(client).await?; Ok(()) } diff --git a/crates/core/src/schema/v1_0_4.rs b/crates/core/src/schema/v1_0_4.rs new file mode 100644 index 00000000..738b0052 --- /dev/null +++ b/crates/core/src/schema/v1_0_4.rs @@ -0,0 +1,37 @@ +use crate::postgres::{PostgresClient, PostgresError}; + +const SCHEMA_SQL: &str = r#" + CREATE INDEX IF NOT EXISTS idx_transaction_audit_attempt_lookup + ON relayer.transaction_audit_log(id, history_id) + WHERE hash IS NOT NULL AND sent_at IS NOT NULL; + + DROP INDEX IF EXISTS relayer.idx_relayer_live_normal_wallet_namespace; + + CREATE UNIQUE INDEX IF NOT EXISTS idx_relayer_live_normal_root_wallet_namespace + ON relayer.record(chain_id, wallet_index) + WHERE deleted = FALSE + AND is_private_key = FALSE + AND wallet_index >= 0 + AND cloned_from_chain_id IS NULL; +"#; + +/// Adds transaction integrity indexes for ordered attempt recovery and stable +/// live normal-wallet namespaces. Both indexes are idempotent. Cloned rows are +/// aliases for an existing signing identity, so the root-identity uniqueness +/// constraint deliberately excludes them. +pub async fn apply_v1_0_4_schema(client: &PostgresClient) -> Result<(), PostgresError> { + client.batch_execute(SCHEMA_SQL).await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::SCHEMA_SQL; + + #[test] + fn live_root_wallet_indexes_are_unique_without_rejecting_canonical_clones() { + assert!(SCHEMA_SQL.contains("CREATE UNIQUE INDEX")); + assert!(SCHEMA_SQL.contains("wallet_index >= 0")); + assert!(SCHEMA_SQL.contains("cloned_from_chain_id IS NULL")); + } +} diff --git a/crates/core/src/transaction/api/send_random_transaction.rs b/crates/core/src/transaction/api/send_random_transaction.rs index 55fc8b54..54dab32e 100644 --- a/crates/core/src/transaction/api/send_random_transaction.rs +++ b/crates/core/src/transaction/api/send_random_transaction.rs @@ -61,3 +61,48 @@ async fn select_random_relayer( )) }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::transaction::types::{TransactionHash, TransactionId}; + use axum::{body::to_bytes, http::StatusCode, response::IntoResponse}; + use serde_json::json; + use std::str::FromStr; + + #[tokio::test] + async fn random_submission_returns_http_200_with_explicit_null_hash_when_pending() { + let id = TransactionId::from_str("11111111-1111-4111-8111-111111111111").unwrap(); + let response = Json(SendTransactionResult { id, hash: None }).into_response(); + + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({ + "id": "11111111-1111-4111-8111-111111111111", + "hash": null + }) + ); + } + + #[tokio::test] + async fn random_submission_returns_http_200_with_known_hash() { + let id = TransactionId::from_str("11111111-1111-4111-8111-111111111111").unwrap(); + let hash = TransactionHash::from_str( + "0x2222222222222222222222222222222222222222222222222222222222222222", + ) + .unwrap(); + let response = Json(SendTransactionResult { id, hash: Some(hash) }).into_response(); + + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({ + "id": "11111111-1111-4111-8111-111111111111", + "hash": "0x2222222222222222222222222222222222222222222222222222222222222222" + }) + ); + } +} diff --git a/crates/core/src/transaction/api/send_transaction.rs b/crates/core/src/transaction/api/send_transaction.rs index 3abce884..5458fbb9 100644 --- a/crates/core/src/transaction/api/send_transaction.rs +++ b/crates/core/src/transaction/api/send_transaction.rs @@ -48,7 +48,7 @@ impl FromStr for RelayTransactionRequest { #[derive(Debug, Serialize, Deserialize)] pub struct SendTransactionResult { pub id: TransactionId, - pub hash: TransactionHash, + pub hash: Option, } /// API endpoint to send a new transaction through a relayer. @@ -127,12 +127,8 @@ pub async fn send_transaction( .add_transaction(&relayer.id, &transaction_to_send) .await?; - let result = SendTransactionResult { - id: transaction.id, - hash: transaction.known_transaction_hash.ok_or(internal_server_error(Some( - "should always have a known transaction hash".to_string(), - )))?, - }; + let result = + SendTransactionResult { id: transaction.id, hash: transaction.known_transaction_hash }; if let Some(reservation) = rate_limit_reservation { reservation.commit(); @@ -140,3 +136,50 @@ pub async fn send_transaction( Ok(result) } + +#[cfg(test)] +mod tests { + use super::*; + use axum::{body::to_bytes, http::StatusCode, response::IntoResponse}; + use serde_json::json; + + fn transaction_id() -> TransactionId { + TransactionId::from_str("11111111-1111-4111-8111-111111111111").unwrap() + } + + #[tokio::test] + async fn direct_submission_returns_http_200_with_explicit_null_hash_when_pending() { + let response = + Json(SendTransactionResult { id: transaction_id(), hash: None }).into_response(); + + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({ + "id": "11111111-1111-4111-8111-111111111111", + "hash": null + }) + ); + } + + #[tokio::test] + async fn direct_submission_returns_http_200_with_known_hash() { + let hash = TransactionHash::from_str( + "0x2222222222222222222222222222222222222222222222222222222222222222", + ) + .unwrap(); + let response = + Json(SendTransactionResult { id: transaction_id(), hash: Some(hash) }).into_response(); + + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({ + "id": "11111111-1111-4111-8111-111111111111", + "hash": "0x2222222222222222222222222222222222222222222222222222222222222222" + }) + ); + } +} diff --git a/crates/core/src/transaction/db/mod.rs b/crates/core/src/transaction/db/mod.rs index f295697b..5051d998 100644 --- a/crates/core/src/transaction/db/mod.rs +++ b/crates/core/src/transaction/db/mod.rs @@ -1,3 +1,18 @@ mod builders; mod read; mod write; + +use crate::{ + gas::{BlobGasPriceResult, GasPriceResult}, + transaction::types::TransactionHash, +}; + +/// One durable broadcast candidate reconstructed from the transaction's live +/// row and append-only audit history. Transaction identity, nonce and relayer +/// remain on the parent transaction record. +#[derive(Clone, Debug)] +pub(crate) struct RecordedTransactionAttempt { + pub(crate) hash: TransactionHash, + pub(crate) sent_with_gas: Option, + pub(crate) sent_with_blob_gas: Option, +} diff --git a/crates/core/src/transaction/db/read.rs b/crates/core/src/transaction/db/read.rs index 4198e3d2..6abea372 100644 --- a/crates/core/src/transaction/db/read.rs +++ b/crates/core/src/transaction/db/read.rs @@ -1,4 +1,4 @@ -use super::builders::build_transaction_from_transaction_view; +use super::{builders::build_transaction_from_transaction_view, RecordedTransactionAttempt}; use crate::{ postgres::{PostgresClient, PostgresError}, relayer::RelayerId, @@ -7,6 +7,72 @@ use crate::{ }; impl PostgresClient { + /// Loads unique broadcast attempts in durable persistence order. + /// + /// Audit history supplies the stable sequence. The live row is a fallback + /// for installations that predate attempt snapshots or were interrupted + /// between the live-row update and an older audit-writing path. + pub(crate) async fn get_recorded_transaction_attempts( + &self, + id: &TransactionId, + ) -> Result, PostgresError> { + let rows = self + .query( + " + WITH raw_attempts AS ( + SELECT + hash, + sent_with_gas, + sent_with_blob_gas, + sent_at, + NULL::BIGINT AS history_id + FROM relayer.transaction + WHERE id = $1 + + UNION ALL + + SELECT + hash, + sent_with_gas, + sent_with_blob_gas, + sent_at, + history_id::BIGINT + FROM relayer.transaction_audit_log + WHERE id = $1 + ), unique_attempts AS ( + SELECT DISTINCT ON (hash) + hash, + sent_with_gas, + sent_with_blob_gas, + sent_at, + history_id + FROM raw_attempts + WHERE hash IS NOT NULL + AND sent_at IS NOT NULL + ORDER BY hash, history_id ASC NULLS LAST + ) + SELECT hash, sent_with_gas, sent_with_blob_gas + FROM unique_attempts + ORDER BY history_id ASC NULLS LAST, sent_at ASC, hash ASC; + ", + &[id], + ) + .await?; + + Ok(rows + .iter() + .map(|row| RecordedTransactionAttempt { + hash: row.get("hash"), + sent_with_gas: row + .get::<_, Option>("sent_with_gas") + .and_then(|value| serde_json::from_value(value).ok()), + sent_with_blob_gas: row + .get::<_, Option>("sent_with_blob_gas") + .and_then(|value| serde_json::from_value(value).ok()), + }) + .collect()) + } + pub async fn get_transaction( &self, id: &TransactionId, @@ -83,6 +149,41 @@ impl PostgresClient { Ok(PagingResult::new(results, paging_context.next(result_count), paging_context.previous())) } + pub async fn get_pending_transactions_with_attempt_evidence_for_relayer( + &self, + id: &RelayerId, + paging_context: &PagingContext, + ) -> Result, PostgresError> { + let rows = self + .query( + " + SELECT * + FROM relayer.transaction + WHERE relayer_id = $1 + AND status = $2 + AND sent_at IS NOT NULL + AND failed_at IS NULL + AND hash IS NOT NULL + ORDER BY nonce ASC + LIMIT $3 + OFFSET $4; + ", + &[ + id, + &TransactionStatus::PENDING, + &(paging_context.limit as i64), + &(paging_context.offset as i64), + ], + ) + .await?; + + let results: Vec = + rows.iter().map(build_transaction_from_transaction_view).collect(); + let result_count = results.len(); + + Ok(PagingResult::new(results, paging_context.next(result_count), paging_context.previous())) + } + pub async fn get_transaction_by_hash( &self, hash: &TransactionHash, diff --git a/crates/core/src/transaction/db/write.rs b/crates/core/src/transaction/db/write.rs index 1c293e32..238e9510 100644 --- a/crates/core/src/transaction/db/write.rs +++ b/crates/core/src/transaction/db/write.rs @@ -17,6 +17,20 @@ use serde_json; const TRANSACTION_TABLES: [&str; 2] = ["relayer.transaction", "relayer.transaction_audit_log"]; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RepairedPoisonedPendingTransactions { + pub count: i64, + pub min_nonce: Option, + pub max_nonce: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RepairedAbsentFutureNoncePendingTransactions { + pub count: i64, + pub min_nonce: Option, + pub max_nonce: Option, +} + impl PostgresClient { pub async fn save_transaction( &mut self, @@ -59,6 +73,124 @@ impl PostgresClient { Ok(()) } + /// Persists the exact signed broadcast candidate before its bytes may be + /// handed to an RPC endpoint. The live transaction row and its audit + /// snapshot together form the recovery record; no separate attempt table + /// is required. + pub async fn transaction_broadcast_attempt( + &mut self, + relayer_id: &RelayerId, + transaction: &Transaction, + transaction_hash: &TransactionHash, + sent_with_gas: &GasPriceResult, + sent_with_blob_gas: Option<&BlobGasPriceResult>, + legacy_transaction: bool, + ) -> Result<(), PostgresError> { + let mut conn = self.pool.get().await?; + let trans = conn.transaction().await.map_err(PostgresError::PgError)?; + + let max_priority_fee_option = + option_if(!legacy_transaction, &sent_with_gas.max_priority_fee); + let max_fee_option = option_if(!legacy_transaction, &sent_with_gas.max_fee); + let legacy_gas_price = option_if(legacy_transaction, sent_with_gas.legacy_gas_price()); + let sent_with_gas_json = + serde_json::to_value(sent_with_gas).unwrap_or(serde_json::Value::Null); + let sent_with_blob_gas_json = sent_with_blob_gas + .map(|blob_gas| serde_json::to_value(blob_gas).unwrap_or(serde_json::Value::Null)); + + trans + .execute( + " + INSERT INTO relayer.transaction ( + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, hash, external_id, + cancelled_by_transaction_id, sent_max_priority_fee_per_gas, + sent_max_fee_per_gas, gas_price, sent_with_gas, sent_with_blob_gas, sent_at + ) + VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, + $16, $17, $18, $19, $20, $21, $22, NOW() + ) + ON CONFLICT (id) DO UPDATE + SET relayer_id = EXCLUDED.relayer_id, + \"to\" = EXCLUDED.\"to\", + \"from\" = EXCLUDED.\"from\", + nonce = EXCLUDED.nonce, + chain_id = EXCLUDED.chain_id, + data = EXCLUDED.data, + value = EXCLUDED.value, + blobs = EXCLUDED.blobs, + gas_limit = EXCLUDED.gas_limit, + speed = EXCLUDED.speed, + status = EXCLUDED.status, + expires_at = EXCLUDED.expires_at, + queued_at = EXCLUDED.queued_at, + hash = EXCLUDED.hash, + external_id = EXCLUDED.external_id, + cancelled_by_transaction_id = EXCLUDED.cancelled_by_transaction_id, + sent_max_priority_fee_per_gas = EXCLUDED.sent_max_priority_fee_per_gas, + sent_max_fee_per_gas = EXCLUDED.sent_max_fee_per_gas, + gas_price = EXCLUDED.gas_price, + sent_with_gas = EXCLUDED.sent_with_gas, + sent_with_blob_gas = EXCLUDED.sent_with_blob_gas, + sent_at = EXCLUDED.sent_at; + ", + &[ + &transaction.id, + relayer_id, + &transaction.to, + &transaction.from, + &transaction.nonce, + &transaction.chain_id, + &transaction.data, + &transaction.value, + &transaction.blobs, + &transaction.gas_limit, + &transaction.speed, + &transaction.status, + &transaction.expires_at, + &transaction.queued_at, + transaction_hash, + &transaction.external_id, + &transaction.cancelled_by_transaction_id, + &max_priority_fee_option, + &max_fee_option, + &legacy_gas_price, + &sent_with_gas_json, + &sent_with_blob_gas_json, + ], + ) + .await?; + + trans + .execute( + " + INSERT INTO relayer.transaction_audit_log ( + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, sent_at, mined_at, + confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, expired_at, + external_id, cancelled_by_transaction_id + ) + SELECT + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, sent_at, mined_at, + confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, expired_at, + external_id, cancelled_by_transaction_id + FROM relayer.transaction + WHERE id = $1; + ", + &[&transaction.id], + ) + .await?; + + trans.commit().await?; + Ok(()) + } + pub async fn transaction_sent( &mut self, transaction_id: &TransactionId, @@ -142,6 +274,268 @@ impl PostgresClient { Ok(()) } + /// Advances a landed, previously recorded attempt from PENDING to INMEMPOOL. + /// The guarded update and audit snapshot commit before startup loads any + /// in-memory queue state. + pub async fn recover_landed_transaction_attempt( + &self, + relayer_id: &RelayerId, + transaction_id: &TransactionId, + transaction_hash: &TransactionHash, + sent_with_gas: &GasPriceResult, + sent_with_blob_gas: Option<&BlobGasPriceResult>, + legacy_transaction: bool, + ) -> Result { + let mut conn = self.pool.get().await?; + let trans = conn.transaction().await.map_err(PostgresError::PgError)?; + + let max_priority_fee_option = + option_if(!legacy_transaction, &sent_with_gas.max_priority_fee); + let max_fee_option = option_if(!legacy_transaction, &sent_with_gas.max_fee); + let legacy_gas_price = option_if(legacy_transaction, sent_with_gas.legacy_gas_price()); + let sent_with_gas_json = + serde_json::to_value(sent_with_gas).unwrap_or(serde_json::Value::Null); + let sent_with_blob_gas_json = sent_with_blob_gas + .map(|blob_gas| serde_json::to_value(blob_gas).unwrap_or(serde_json::Value::Null)); + + let row = trans + .query_one( + " + WITH recovered AS ( + UPDATE relayer.transaction + SET status = $4, + hash = $3, + sent_max_priority_fee_per_gas = $5, + sent_max_fee_per_gas = $6, + gas_price = $7, + sent_with_gas = $8, + sent_with_blob_gas = $9, + sent_at = COALESCE(sent_at, NOW()) + WHERE id = $1 + AND relayer_id = $2 + AND status = $10 + AND failed_at IS NULL + AND EXISTS ( + SELECT 1 + FROM relayer.transaction_audit_log attempt + WHERE attempt.id = $1 + AND attempt.hash = $3 + AND attempt.sent_at IS NOT NULL + ) + RETURNING * + ), audit AS ( + INSERT INTO relayer.transaction_audit_log ( + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, sent_at, mined_at, + confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, expired_at, + external_id, cancelled_by_transaction_id + ) + SELECT + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, sent_at, mined_at, + confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, expired_at, + external_id, cancelled_by_transaction_id + FROM recovered + RETURNING id + ) + SELECT COUNT(*)::BIGINT AS recovered_count FROM audit; + ", + &[ + transaction_id, + relayer_id, + transaction_hash, + &TransactionStatus::INMEMPOOL, + &max_priority_fee_option, + &max_fee_option, + &legacy_gas_price, + &sent_with_gas_json, + &sent_with_blob_gas_json, + &TransactionStatus::PENDING, + ], + ) + .await + .map_err(PostgresError::PgError)?; + + trans.commit().await.map_err(PostgresError::PgError)?; + Ok(row.get::<_, i64>("recovered_count") == 1) + } + + pub async fn repair_poisoned_pending_transactions_for_relayer( + &self, + relayer_id: &RelayerId, + ) -> Result { + let mut conn = self.pool.get().await?; + let trans = conn.transaction().await.map_err(PostgresError::PgError)?; + + let row = trans + .query_one( + " + WITH repaired AS ( + UPDATE relayer.transaction + SET status = $2, + failed_reason = COALESCE(failed_reason, $4) + WHERE relayer_id = $1 + AND status = $3 + AND failed_at IS NOT NULL + AND hash IS NULL + AND sent_at IS NULL + RETURNING * + ), audit AS ( + INSERT INTO relayer.transaction_audit_log ( + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, sent_at, mined_at, + confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, expired_at, + external_id, cancelled_by_transaction_id + ) + SELECT + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, + gas_limit, speed, status, expires_at, queued_at, sent_at, mined_at, + confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, expired_at, + external_id, cancelled_by_transaction_id + FROM repaired + RETURNING nonce + ) + SELECT COUNT(*)::BIGINT AS repaired_count, + MIN(nonce)::BIGINT AS min_nonce, + MAX(nonce)::BIGINT AS max_nonce + FROM audit; + ", + &[ + relayer_id, + &TransactionStatus::FAILED, + &TransactionStatus::PENDING, + &"startup repair: terminalized unsent failed pending transaction", + ], + ) + .await + .map_err(PostgresError::PgError)?; + + trans.commit().await.map_err(PostgresError::PgError)?; + + Ok(RepairedPoisonedPendingTransactions { + count: row.get("repaired_count"), + min_nonce: row.get::<_, Option>("min_nonce").map(TransactionNonce::from), + max_nonce: row.get::<_, Option>("max_nonce").map(TransactionNonce::from), + }) + } + + pub async fn repair_absent_future_nonce_pending_transactions_for_relayer( + &self, + relayer_id: &RelayerId, + chain_nonce: &TransactionNonce, + checked_absent_transactions: &[(TransactionId, Vec)], + failed_reason: &str, + ) -> Result { + if checked_absent_transactions.is_empty() { + return Ok(RepairedAbsentFutureNoncePendingTransactions { + count: 0, + min_nonce: None, + max_nonce: None, + }); + } + + let mut conn = self.pool.get().await?; + let trans = conn.transaction().await.map_err(PostgresError::PgError)?; + + let mut count = 0; + let mut min_nonce: Option = None; + let mut max_nonce: Option = None; + + for (transaction_id, transaction_hashes) in checked_absent_transactions { + let row = trans + .query_one( + " + WITH repaired AS ( + UPDATE relayer.transaction + SET status = $3, + failed_at = NOW(), + failed_reason = COALESCE(failed_reason, $5) + WHERE id = $1 + AND relayer_id = $2 + AND status = $4 + AND sent_at IS NOT NULL + AND failed_at IS NULL + AND hash IS NOT NULL + AND hash = ANY($7::BYTEA[]) + AND NOT EXISTS ( + SELECT 1 + FROM relayer.transaction_audit_log attempts + WHERE attempts.id = $1 + AND attempts.sent_at IS NOT NULL + AND attempts.hash IS NOT NULL + AND NOT (attempts.hash = ANY($7::BYTEA[])) + ) + AND nonce > $6 + RETURNING * + ), audit AS ( + INSERT INTO relayer.transaction_audit_log ( + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, + blobs, gas_limit, speed, status, expires_at, queued_at, sent_at, + mined_at, confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, + expired_at, external_id, cancelled_by_transaction_id + ) + SELECT + id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, + blobs, gas_limit, speed, status, expires_at, queued_at, sent_at, + mined_at, confirmed_at, failed_at, failed_reason, hash, + sent_max_priority_fee_per_gas, sent_max_fee_per_gas, gas_price, + sent_with_gas, sent_with_blob_gas, block_hash, block_number, + expired_at, external_id, cancelled_by_transaction_id + FROM repaired + RETURNING nonce + ) + SELECT COUNT(*)::BIGINT AS repaired_count, + MIN(nonce)::BIGINT AS min_nonce, + MAX(nonce)::BIGINT AS max_nonce + FROM audit; + ", + &[ + transaction_id, + relayer_id, + &TransactionStatus::FAILED, + &TransactionStatus::PENDING, + &failed_reason, + chain_nonce, + transaction_hashes, + ], + ) + .await + .map_err(PostgresError::PgError)?; + + let repaired_count: i64 = row.get("repaired_count"); + count += repaired_count; + + if let Some(nonce) = row.get::<_, Option>("min_nonce").map(TransactionNonce::from) + { + min_nonce = Some(match min_nonce { + Some(current) if current.into_inner() <= nonce.into_inner() => current, + _ => nonce, + }); + } + if let Some(nonce) = row.get::<_, Option>("max_nonce").map(TransactionNonce::from) + { + max_nonce = Some(match max_nonce { + Some(current) if current.into_inner() >= nonce.into_inner() => current, + _ => nonce, + }); + } + } + + trans.commit().await.map_err(PostgresError::PgError)?; + + Ok(RepairedAbsentFutureNoncePendingTransactions { count, min_nonce, max_nonce }) + } + pub async fn transaction_failed_on_send( &self, relayer_id: &RelayerId, @@ -488,50 +882,6 @@ impl PostgresClient { Ok(()) } - /// Records the hash of a signed payload whose broadcast outcome is unknown (the - /// transaction row stays PENDING). Restores the ability to recognise the broadcast - /// as our own if it mines, even across a restart. - pub async fn transaction_update_known_hash( - &mut self, - transaction_id: &TransactionId, - transaction_hash: &TransactionHash, - ) -> Result<(), PostgresError> { - let mut conn = self.pool.get().await?; - let trans = conn.transaction().await.map_err(PostgresError::PgError)?; - - trans - .execute( - "UPDATE relayer.transaction SET hash = $2 WHERE id = $1", - &[&transaction_id, &transaction_hash], - ) - .await?; - - trans - .execute( - " - INSERT INTO relayer.transaction_audit_log ( - id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, gas_limit, - speed, status, expires_at, queued_at, sent_at, mined_at, confirmed_at, - failed_at, failed_reason, hash, sent_max_priority_fee_per_gas, - sent_max_fee_per_gas, gas_price, block_hash, block_number, external_id - ) - SELECT - id, relayer_id, \"to\", \"from\", nonce, chain_id, data, value, blobs, gas_limit, - speed, status, expires_at, queued_at, sent_at, mined_at, confirmed_at, - failed_at, failed_reason, $2, sent_max_priority_fee_per_gas, - sent_max_fee_per_gas, gas_price, block_hash, block_number, external_id - FROM relayer.transaction - WHERE id = $1; - ", - &[&transaction_id, &transaction_hash], - ) - .await?; - - trans.commit().await?; - - Ok(()) - } - pub async fn transaction_expired( &mut self, transaction_id: &TransactionId, diff --git a/crates/core/src/transaction/nonce_manager.rs b/crates/core/src/transaction/nonce_manager.rs index 2dc40ddb..033228a2 100644 --- a/crates/core/src/transaction/nonce_manager.rs +++ b/crates/core/src/transaction/nonce_manager.rs @@ -29,4 +29,79 @@ impl NonceManager { let nonce_guard = self.nonce.lock().await; *nonce_guard } + + pub async fn release_unbroadcast_nonce(&self, nonce: TransactionNonce) { + let mut nonce_guard = self.nonce.lock().await; + if nonce_guard.into_inner() == nonce.into_inner() + 1 { + *nonce_guard = nonce; + } + } + + /// Advances the next-free nonce after the corresponding reservation has + /// already been committed to the database. + pub async fn advance_after_persisted_reservation(&self, nonce: TransactionNonce) { + let mut nonce_guard = self.nonce.lock().await; + let next_nonce = nonce + 1; + if next_nonce.into_inner() > nonce_guard.into_inner() { + *nonce_guard = next_nonce; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + async fn assert_prebroadcast_failure_releases_reservation() { + let nonce_manager = NonceManager::new(TransactionNonce::new(7)); + let reserved_nonce = nonce_manager.get_and_increment().await; + + nonce_manager.release_unbroadcast_nonce(reserved_nonce).await; + + assert_eq!(nonce_manager.get_current_nonce().await, TransactionNonce::new(7)); + assert_eq!(nonce_manager.get_and_increment().await, TransactionNonce::new(7)); + } + + #[tokio::test] + async fn gas_calculation_failure_releases_unbroadcast_nonce() { + assert_prebroadcast_failure_releases_reservation().await; + } + + #[tokio::test] + async fn signing_preparation_failure_releases_unbroadcast_nonce() { + assert_prebroadcast_failure_releases_reservation().await; + } + + #[tokio::test] + async fn database_save_failure_releases_unbroadcast_nonce() { + assert_prebroadcast_failure_releases_reservation().await; + } + + #[tokio::test] + async fn queue_insertion_failure_releases_unbroadcast_nonce() { + assert_prebroadcast_failure_releases_reservation().await; + } + + #[tokio::test] + async fn stale_release_does_not_rewind_a_later_reservation() { + let nonce_manager = NonceManager::new(TransactionNonce::new(7)); + let stale_reserved_nonce = nonce_manager.get_and_increment().await; + let later_reserved_nonce = nonce_manager.get_and_increment().await; + + nonce_manager.release_unbroadcast_nonce(stale_reserved_nonce).await; + + assert_eq!(later_reserved_nonce, TransactionNonce::new(8)); + assert_eq!(nonce_manager.get_current_nonce().await, TransactionNonce::new(9)); + } + + #[tokio::test] + async fn persisted_reservation_advances_next_free_nonce_without_rewinding() { + let nonce_manager = NonceManager::new(TransactionNonce::new(7)); + + nonce_manager.advance_after_persisted_reservation(TransactionNonce::new(9)).await; + assert_eq!(nonce_manager.get_current_nonce().await, TransactionNonce::new(10)); + + nonce_manager.advance_after_persisted_reservation(TransactionNonce::new(8)).await; + assert_eq!(nonce_manager.get_current_nonce().await, TransactionNonce::new(10)); + } } diff --git a/crates/core/src/transaction/queue_system/attempts.rs b/crates/core/src/transaction/queue_system/attempts.rs new file mode 100644 index 00000000..b7e98b69 --- /dev/null +++ b/crates/core/src/transaction/queue_system/attempts.rs @@ -0,0 +1,100 @@ +use async_trait::async_trait; + +use alloy::transports::{RpcError, TransportErrorKind}; + +use crate::{ + provider::EvmProvider, + transaction::{db::RecordedTransactionAttempt, types::TransactionHash}, +}; + +use super::transactions_queue::TransactionsQueue; + +#[async_trait] +pub(super) trait TransactionExistenceChecker { + async fn transaction_exists( + &self, + hash: &TransactionHash, + ) -> Result>; +} + +#[async_trait] +impl TransactionExistenceChecker for EvmProvider { + async fn transaction_exists( + &self, + hash: &TransactionHash, + ) -> Result> { + EvmProvider::transaction_exists(self, hash).await + } +} + +#[async_trait] +impl TransactionExistenceChecker for TransactionsQueue { + async fn transaction_exists( + &self, + hash: &TransactionHash, + ) -> Result> { + TransactionsQueue::transaction_exists(self, hash).await + } +} + +pub(super) async fn find_landed_attempt<'a, C: TransactionExistenceChecker>( + checker: &C, + attempts: &'a [RecordedTransactionAttempt], +) -> Result, RpcError> { + let mut landed = None; + + for attempt in attempts { + if checker.transaction_exists(&attempt.hash).await? && landed.is_none() { + landed = Some(attempt); + } + } + + Ok(landed) +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy::primitives::TxHash; + use std::collections::HashSet; + + struct TestChecker { + landed: HashSet, + } + + #[async_trait] + impl TransactionExistenceChecker for TestChecker { + async fn transaction_exists( + &self, + hash: &TransactionHash, + ) -> Result> { + Ok(self.landed.contains(hash)) + } + } + + fn attempt(byte: u8) -> RecordedTransactionAttempt { + RecordedTransactionAttempt { + hash: TransactionHash::new(TxHash::repeat_byte(byte)), + sent_with_gas: None, + sent_with_blob_gas: None, + } + } + + #[tokio::test] + async fn recovery_returns_first_landed_attempt_in_recorded_order() { + let attempts = vec![attempt(1), attempt(2), attempt(3)]; + let checker = TestChecker { landed: HashSet::from([attempts[1].hash, attempts[2].hash]) }; + + let landed = find_landed_attempt(&checker, &attempts).await.unwrap().unwrap(); + + assert_eq!(landed.hash, attempts[1].hash); + } + + #[tokio::test] + async fn recovery_returns_none_only_when_every_recorded_attempt_is_absent() { + let attempts = vec![attempt(1), attempt(2)]; + let checker = TestChecker { landed: HashSet::new() }; + + assert!(find_landed_attempt(&checker, &attempts).await.unwrap().is_none()); + } +} diff --git a/crates/core/src/transaction/queue_system/mod.rs b/crates/core/src/transaction/queue_system/mod.rs index 517394be..5949027a 100644 --- a/crates/core/src/transaction/queue_system/mod.rs +++ b/crates/core/src/transaction/queue_system/mod.rs @@ -5,5 +5,6 @@ pub use transactions_queues::TransactionsQueues; mod types; pub use types::{ReplaceTransactionResult, TransactionToSend, TransactionsQueueSetup}; +mod attempts; mod start; pub use start::{startup_transactions_queues, StartTransactionsQueuesError}; diff --git a/crates/core/src/transaction/queue_system/start.rs b/crates/core/src/transaction/queue_system/start.rs index a1feb9a2..a7e559c6 100644 --- a/crates/core/src/transaction/queue_system/start.rs +++ b/crates/core/src/transaction/queue_system/start.rs @@ -1,10 +1,14 @@ use std::{collections::VecDeque, sync::Arc}; +use alloy::transports::{RpcError, TransportErrorKind}; use thiserror::Error; use tokio::sync::Mutex; use tracing::{error, info, warn}; -use super::{transactions_queues::TransactionsQueues, types::TransactionRelayerSetup}; +use super::{ + attempts::find_landed_attempt, transactions_queues::TransactionsQueues, + types::TransactionRelayerSetup, +}; use crate::transaction::queue_system::types::{ CompetitionType, CompetitiveTransaction, ProcessInmempoolTransactionError, ProcessMinedTransactionError, ProcessPendingTransactionError, @@ -22,7 +26,7 @@ use crate::{ utils::sleep_ms, }, shutdown::subscribe_to_shutdown, - transaction::types::{Transaction, TransactionStatus}, + transaction::types::{Transaction, TransactionId, TransactionNonce, TransactionStatus}, }; pub async fn spawn_processing_tasks_for_relayer( @@ -339,6 +343,187 @@ async fn repopulate_competitive_transaction_queue( Ok(competitive_queue) } +pub(super) fn effective_startup_nonce( + onchain_nonce: TransactionNonce, + pending_transactions: &VecDeque, + inmempool_transactions: &VecDeque, +) -> TransactionNonce { + let pending_max = + pending_transactions.iter().map(|transaction| transaction.nonce.into_inner()).max(); + let inmempool_max = inmempool_transactions + .iter() + .map(|transaction| { + let original_nonce = transaction.original.nonce.into_inner(); + transaction + .competitive + .as_ref() + .map(|(competitor, _)| original_nonce.max(competitor.nonce.into_inner())) + .unwrap_or(original_nonce) + }) + .max(); + + let next_after_actionable = pending_max + .into_iter() + .chain(inmempool_max) + .max() + .map(|nonce| nonce.saturating_add(1)) + .unwrap_or(0); + + TransactionNonce::new(onchain_nonce.into_inner().max(next_after_actionable)) +} + +pub(super) fn future_nonce_pending_repair_hash( + transaction: &Transaction, + chain_nonce: TransactionNonce, +) -> Option { + if transaction.status == TransactionStatus::PENDING + && transaction.sent_at.is_some() + && transaction.nonce.into_inner() > chain_nonce.into_inner() + { + transaction.known_transaction_hash + } else { + None + } +} + +#[derive(Error, Debug)] +enum RepairRecordedAttemptsError { + #[error("failed to get chain nonce for relayer {0}: {1}")] + CouldNotGetChainNonce(RelayerId, WalletOrProviderError), + + #[error("failed to load candidate transactions for relayer {0}: {1}")] + CouldNotLoadCandidates(RelayerId, PostgresError), + + #[error("failed to load recorded attempts for transaction {1} on relayer {0}: {2}")] + CouldNotLoadAttempts(RelayerId, TransactionId, PostgresError), + + #[error("transaction {1} on relayer {0} has broadcast evidence but no recorded attempts")] + MissingAttempts(RelayerId, TransactionId), + + #[error("failed to check recorded attempts for transaction {1} on relayer {0}: {2}")] + CouldNotCheckAttempts(RelayerId, TransactionId, RpcError), + + #[error("landed attempt {2} for transaction {1} on relayer {0} is missing gas evidence")] + MissingGasEvidence(RelayerId, TransactionId, crate::transaction::types::TransactionHash), + + #[error("failed to persist landed attempt recovery for transaction {1} on relayer {0}: {2}")] + CouldNotRecoverLandedAttempt(RelayerId, TransactionId, PostgresError), + + #[error("transaction {1} on relayer {0} changed before its landed attempt could be recovered")] + LandedAttemptStateChanged(RelayerId, TransactionId), + + #[error("failed to terminalize broadcast-absent transactions for relayer {0}: {1}")] + CouldNotRepairAbsentAttempts(RelayerId, PostgresError), +} + +async fn repair_recorded_attempts_for_relayer( + db: &PostgresClient, + relayer: &Relayer, + provider: &EvmProvider, +) -> Result<(), RepairRecordedAttemptsError> { + let chain_nonce = provider + .get_nonce(relayer) + .await + .map_err(|error| RepairRecordedAttemptsError::CouldNotGetChainNonce(relayer.id, error))?; + + let mut candidates = Vec::new(); + let mut paging_context = PagingContext::new(1000, 0); + loop { + let results = db + .get_pending_transactions_with_attempt_evidence_for_relayer( + &relayer.id, + &paging_context, + ) + .await + .map_err(|error| { + RepairRecordedAttemptsError::CouldNotLoadCandidates(relayer.id, error) + })?; + let result_count = results.items.len(); + candidates.extend(results.items); + + match paging_context.next(result_count) { + Some(next) => paging_context = next, + None => break, + } + } + + let mut absent = Vec::new(); + + for transaction in candidates { + let attempts = + db.get_recorded_transaction_attempts(&transaction.id).await.map_err(|error| { + RepairRecordedAttemptsError::CouldNotLoadAttempts(relayer.id, transaction.id, error) + })?; + + if attempts.is_empty() { + return Err(RepairRecordedAttemptsError::MissingAttempts(relayer.id, transaction.id)); + } + + let landed = find_landed_attempt(provider, &attempts).await.map_err(|error| { + RepairRecordedAttemptsError::CouldNotCheckAttempts(relayer.id, transaction.id, error) + })?; + + if let Some(attempt) = landed { + let sent_with_gas = attempt.sent_with_gas.as_ref().ok_or( + RepairRecordedAttemptsError::MissingGasEvidence( + relayer.id, + transaction.id, + attempt.hash, + ), + )?; + let recovered = db + .recover_landed_transaction_attempt( + &relayer.id, + &transaction.id, + &attempt.hash, + sent_with_gas, + attempt.sent_with_blob_gas.as_ref(), + !relayer.eip_1559_enabled, + ) + .await + .map_err(|error| { + RepairRecordedAttemptsError::CouldNotRecoverLandedAttempt( + relayer.id, + transaction.id, + error, + ) + })?; + if !recovered { + return Err(RepairRecordedAttemptsError::LandedAttemptStateChanged( + relayer.id, + transaction.id, + )); + } + } else if future_nonce_pending_repair_hash(&transaction, chain_nonce).is_some() { + absent.push((transaction.id, attempts.iter().map(|attempt| attempt.hash).collect())); + } + } + + let repair_result = db + .repair_absent_future_nonce_pending_transactions_for_relayer( + &relayer.id, + &chain_nonce, + &absent, + "startup repair: terminalized broadcast-absent future-nonce pending transaction", + ) + .await + .map_err(|error| { + RepairRecordedAttemptsError::CouldNotRepairAbsentAttempts(relayer.id, error) + })?; + + info!( + "Startup recorded-attempt reconciliation for relayer {} ({}): chain nonce {}, absent repairs {}, nonce range {:?}..={:?}", + relayer.name, + relayer.id, + chain_nonce.into_inner(), + repair_result.count, + repair_result.min_nonce.map(|nonce| nonce.into_inner()), + repair_result.max_nonce.map(|nonce| nonce.into_inner()) + ); + + Ok(()) +} + /// Loads all relayers from the database. async fn load_relayers(db: &PostgresClient) -> Result, PostgresError> { let mut relayers: Vec = Vec::new(); @@ -531,6 +716,50 @@ pub async fn startup_transactions_queues( let relayer_id = relayer.id; + let poisoned_repair = match postgres + .repair_poisoned_pending_transactions_for_relayer(&relayer_id) + .await + { + Ok(result) => result, + Err(error) => { + error!( + "Startup is keeping relayer {} ({}) offline because poisoned pending repair could not be persisted: {}", + relayer.name, relayer_id, error + ); + continue; + } + }; + + if poisoned_repair.count > 0 { + info!( + "Startup repaired poisoned unsent pending transactions for relayer {} ({}): count {}, nonce range {:?}..={:?}", + relayer.name, + relayer_id, + poisoned_repair.count, + poisoned_repair.min_nonce.map(|nonce| nonce.into_inner()), + poisoned_repair.max_nonce.map(|nonce| nonce.into_inner()) + ); + } + + if let Err(error) = + repair_recorded_attempts_for_relayer(&postgres, &relayer, &evm_provider).await + { + error!( + "Startup is keeping relayer {} ({}) offline because recorded-attempt recovery is unresolved: {}", + relayer.name, relayer_id, error + ); + continue; + } + + let pending_transactions = repopulate_transaction_queue( + &postgres, + &relayer_id, + &TransactionStatus::PENDING, + ) + .await?; + let inmempool_transactions = + repopulate_competitive_transaction_queue(&postgres, &relayer_id).await?; + let mined_transactions = repopulate_transaction_queue(&postgres, &relayer_id, &TransactionStatus::MINED) .await?; @@ -548,13 +777,8 @@ pub async fn startup_transactions_queues( transaction_relayer_setups.push(TransactionRelayerSetup::new( relayer, evm_provider, - repopulate_transaction_queue( - &postgres, - &relayer_id, - &TransactionStatus::PENDING, - ) - .await?, - repopulate_competitive_transaction_queue(&postgres, &relayer_id).await?, + pending_transactions, + inmempool_transactions, mined_transactions .into_iter() .map(|transaction| (transaction.id, transaction)) @@ -582,3 +806,77 @@ pub async fn startup_transactions_queues( Ok(transactions_queues) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + gas::GasLimit, + network::ChainId, + shared::common_types::EvmAddress, + transaction::types::{ + TransactionData, TransactionHash, TransactionId, TransactionSpeed, TransactionValue, + }, + }; + use alloy::primitives::TxHash; + use chrono::Utc; + + fn pending_transaction(nonce: u64) -> Transaction { + let now = Utc::now(); + Transaction { + id: TransactionId::new(), + relayer_id: RelayerId::new(), + to: EvmAddress::zero(), + from: EvmAddress::zero(), + value: TransactionValue::zero(), + data: TransactionData::empty(), + nonce: TransactionNonce::new(nonce), + chain_id: ChainId::new(1), + gas_limit: Some(GasLimit::new(21_000)), + status: TransactionStatus::PENDING, + blobs: None, + known_transaction_hash: None, + queued_at: now, + expires_at: now, + sent_at: None, + confirmed_at: None, + sent_with_gas: None, + sent_with_blob_gas: None, + mined_at: None, + mined_at_block_number: None, + speed: TransactionSpeed::FAST, + sent_with_max_priority_fee_per_gas: None, + sent_with_max_fee_per_gas: None, + is_noop: false, + external_id: None, + cancelled_by_transaction_id: None, + failed_reason: None, + } + } + + #[test] + fn startup_nonce_protects_actionable_pending_rows_above_provider_nonce() { + let pending = VecDeque::from([pending_transaction(10)]); + + let nonce = effective_startup_nonce(TransactionNonce::new(7), &pending, &VecDeque::new()); + + assert_eq!(nonce, TransactionNonce::new(11)); + } + + #[test] + fn future_nonce_terminalization_requires_complete_attempt_markers() { + let mut transaction = pending_transaction(9); + let hash = TransactionHash::new(TxHash::repeat_byte(9)); + + assert_eq!(future_nonce_pending_repair_hash(&transaction, TransactionNonce::new(7)), None); + + transaction.known_transaction_hash = Some(hash); + transaction.sent_at = Some(Utc::now()); + + assert_eq!( + future_nonce_pending_repair_hash(&transaction, TransactionNonce::new(7)), + Some(hash) + ); + assert_eq!(future_nonce_pending_repair_hash(&transaction, TransactionNonce::new(10)), None); + } +} diff --git a/crates/core/src/transaction/queue_system/transactions_queue.rs b/crates/core/src/transaction/queue_system/transactions_queue.rs index 8ccc41ac..48cbe90e 100644 --- a/crates/core/src/transaction/queue_system/transactions_queue.rs +++ b/crates/core/src/transaction/queue_system/transactions_queue.rs @@ -16,8 +16,8 @@ use crate::{ MaxFee, MaxPriorityFee, BLOB_GAS_PER_BLOB, }, network::ChainId, - postgres::PostgresClient, - provider::{EvmProvider, SendTransactionError}, + postgres::{PostgresClient, PostgresError}, + provider::{EvmProvider, SendTransactionError, SignedTransaction}, relayer::{Relayer, RelayerId}, safe_proxy::SafeProxyManager, shared::common_types::EvmAddress, @@ -27,15 +27,14 @@ use crate::{ types::{Transaction, TransactionHash, TransactionId, TransactionSpeed, TransactionStatus}, }, yaml::GasBumpBlockConfig, - WalletError, }; use alloy::network::{AnyTransactionReceipt, ReceiptResponse}; use alloy::{ - consensus::{SignableTransaction, TypedTransaction}, + consensus::TypedTransaction, hex, - primitives::Signature, transports::{RpcError, TransportErrorKind}, }; +use async_trait::async_trait; use chrono::Utc; use tokio::sync::Mutex; use tracing::error; @@ -125,6 +124,118 @@ fn bump_max_priority_fee_by_at_least_one(max_priority_fee: MaxPriorityFee) -> Ma MaxPriorityFee::new(bump_u128_by_at_least_one(max_priority_fee.into_u128())) } +fn sort_pending_transactions(transactions: &mut VecDeque) { + transactions.make_contiguous().sort_by_key(|transaction| transaction.nonce.into_inner()); +} + +#[async_trait] +trait BroadcastAttemptStore { + async fn persist_attempt( + &mut self, + relayer_id: &RelayerId, + transaction: &Transaction, + transaction_sent: &TransactionSentWithRelayer, + legacy_transaction: bool, + ) -> Result<(), PostgresError>; + + async fn mark_sent( + &mut self, + transaction_sent: &TransactionSentWithRelayer, + legacy_transaction: bool, + ) -> Result<(), PostgresError>; +} + +#[async_trait] +impl BroadcastAttemptStore for PostgresClient { + async fn persist_attempt( + &mut self, + relayer_id: &RelayerId, + transaction: &Transaction, + transaction_sent: &TransactionSentWithRelayer, + legacy_transaction: bool, + ) -> Result<(), PostgresError> { + self.transaction_broadcast_attempt( + relayer_id, + transaction, + &transaction_sent.hash, + &transaction_sent.sent_with_gas, + transaction_sent.sent_with_blob_gas.as_ref(), + legacy_transaction, + ) + .await + } + + async fn mark_sent( + &mut self, + transaction_sent: &TransactionSentWithRelayer, + legacy_transaction: bool, + ) -> Result<(), PostgresError> { + self.transaction_sent( + &transaction_sent.id, + &transaction_sent.hash, + &transaction_sent.sent_with_gas, + transaction_sent.sent_with_blob_gas.as_ref(), + legacy_transaction, + ) + .await + } +} + +#[async_trait] +trait SignedTransactionBroadcaster { + async fn broadcast( + &self, + transaction: &SignedTransaction, + ) -> Result; +} + +#[async_trait] +impl SignedTransactionBroadcaster for EvmProvider { + async fn broadcast( + &self, + transaction: &SignedTransaction, + ) -> Result { + self.send_raw_transaction(transaction).await + } +} + +async fn persist_and_broadcast( + store: &mut S, + broadcaster: &B, + relayer_id: &RelayerId, + transaction: &mut Transaction, + transaction_sent: &TransactionSentWithRelayer, + signed_transaction: &SignedTransaction, + legacy_transaction: bool, +) -> Result<(), TransactionQueueSendTransactionError> { + // No signed bytes may reach an RPC endpoint until the exact hash and gas + // snapshot are durable. + store.persist_attempt(relayer_id, transaction, transaction_sent, legacy_transaction).await?; + + // The persistence above is the source of truth. Only now may memory expose + // the attempt, so an ambiguous send keeps the nonce protected and pollable. + transaction.known_transaction_hash = Some(transaction_sent.hash); + transaction.sent_with_max_fee_per_gas = Some(transaction_sent.sent_with_gas.max_fee); + transaction.sent_with_max_priority_fee_per_gas = + Some(transaction_sent.sent_with_gas.max_priority_fee); + transaction.sent_with_gas = Some(transaction_sent.sent_with_gas.clone()); + transaction.sent_with_blob_gas = transaction_sent.sent_with_blob_gas.clone(); + transaction.sent_at = Some(Utc::now()); + + broadcaster + .broadcast(signed_transaction) + .await + .map_err(TransactionQueueSendTransactionError::TransactionSendError)?; + + // Status advances only after the database has accepted the terminal send + // transition. A failure here leaves the durable attempt in PENDING state for + // evidence-based recovery. + store.mark_sent(transaction_sent, legacy_transaction).await?; + transaction.status = TransactionStatus::INMEMPOOL; + + Ok(()) +} + pub struct TransactionsQueue { pending_transactions: Mutex>, inmempool_transactions: Mutex>, @@ -151,8 +262,10 @@ impl TransactionsQueue { setup.relayer.id, setup.relayer.name, setup.relayer.chain_id ); let confirmations = setup.evm_provider.confirmations; + let mut pending_transactions = setup.pending_transactions; + sort_pending_transactions(&mut pending_transactions); Self { - pending_transactions: Mutex::new(setup.pending_transactions), + pending_transactions: Mutex::new(pending_transactions), inmempool_transactions: Mutex::new(setup.inmempool_transactions), mined_transactions: Mutex::new(setup.mined_transactions), evm_provider: setup.evm_provider, @@ -262,7 +375,11 @@ impl TransactionsQueue { transaction.id, self.relayer.name ); let mut transactions = self.pending_transactions.lock().await; - transactions.push_back(transaction); + let insert_at = transactions + .iter() + .position(|queued| queued.nonce.into_inner() > transaction.nonce.into_inner()) + .unwrap_or(transactions.len()); + transactions.insert(insert_at, transaction); info!( "Pending transactions count for relayer {}: {}", self.relayer.name, @@ -1142,50 +1259,6 @@ impl TransactionsQueue { Ok(blob_gas_price) } - pub async fn compute_tx_hash( - &self, - transaction: &TypedTransaction, - ) -> Result { - info!("Computing transaction hash for relayer: {}", self.relayer.name); - - let signature = self.evm_provider.sign_transaction(&self.relayer, transaction).await?; - - let tx_hash = Self::signed_transaction_hash(transaction, signature); - info!("Computed transaction hash {} for relayer: {}", tx_hash, self.relayer.name); - Ok(tx_hash) - } - - /// Computes the on-chain hash of an already-signed payload without re-signing. - fn signed_transaction_hash( - transaction: &TypedTransaction, - signature: Signature, - ) -> TransactionHash { - let hash = match transaction { - TypedTransaction::Legacy(tx) => { - let signed = tx.clone().into_signed(signature); - *signed.hash() - } - TypedTransaction::Eip2930(tx) => { - let signed = tx.clone().into_signed(signature); - *signed.hash() - } - TypedTransaction::Eip1559(tx) => { - let signed = tx.clone().into_signed(signature); - *signed.hash() - } - TypedTransaction::Eip4844(tx) => { - let signed = tx.clone().into_signed(signature); - *signed.hash() - } - TypedTransaction::Eip7702(tx) => { - let signed = tx.clone().into_signed(signature); - *signed.hash() - } - }; - - TransactionHash::from_alloy_hash(&hash) - } - pub async fn estimate_gas( &self, transaction_request: &TypedTransaction, @@ -1273,58 +1346,6 @@ impl TransactionsQueue { } } - /// True when the node's send error proves the submitted payload was rejected and is - /// definitively not in the mempool - as opposed to transport failures, where the - /// broadcast may have been accepted with the response lost. - fn send_error_rules_out_broadcast(error_msg: &str) -> bool { - matches!( - classify_send_error(error_msg), - SendErrorClass::InsufficientFunds - | SendErrorClass::PermanentRejection - | SendErrorClass::NonceConflict - | SendErrorClass::Underpriced - ) || error_msg.contains("invalid signature") - } - - /// Records the hash of a signed payload whose broadcast outcome is unknown, in both - /// the in-memory pending entry and the database, so a later 'nonce too low' receipt - /// check can recognise the broadcast as our own instead of reassigning its nonce. - async fn record_broadcast_attempt_hash( - &self, - db: &mut PostgresClient, - transaction: &mut Transaction, - attempt_hash: TransactionHash, - ) { - if transaction.known_transaction_hash == Some(attempt_hash) { - return; - } - - info!( - "Recording broadcast attempt hash {} for transaction {} on relayer: {} (send outcome unknown)", - attempt_hash, transaction.id, self.relayer.name - ); - - transaction.known_transaction_hash = Some(attempt_hash); - - { - let mut transactions = self.pending_transactions.lock().await; - if let Some(stored) = transactions.iter_mut().find(|tx| tx.id == transaction.id) { - stored.known_transaction_hash = Some(attempt_hash); - } - } - - if let Err(db_error) = - db.transaction_update_known_hash(&transaction.id, &attempt_hash).await - { - // In-memory state is already updated; worst case a crash falls back to the - // previously recorded candidate hash - error!( - "Failed to persist broadcast attempt hash for transaction {}: {}", - transaction.id, db_error - ); - } - } - pub async fn send_transaction( &mut self, db: &mut PostgresClient, @@ -1612,14 +1633,11 @@ impl TransactionsQueue { transaction_request, self.relayer.name ); - let mut signature = - self.evm_provider.sign_transaction(&self.relayer, &transaction_request).await.map_err( - |e| { - TransactionQueueSendTransactionError::TransactionSendError( - SendTransactionError::InternalError(e.to_string()), - ) - }, - )?; + let mut signed_transaction = self + .evm_provider + .prepare_signed_transaction(&self.relayer, &transaction_request) + .await + .map_err(TransactionQueueSendTransactionError::TransactionSendError)?; if !transaction.is_noop && can_replace_with_noop && Self::has_expired(transaction) { info!( @@ -1655,94 +1673,36 @@ impl TransactionsQueue { })? }; - signature = self + signed_transaction = self .evm_provider - .sign_transaction(&self.relayer, &transaction_request) + .prepare_signed_transaction(&self.relayer, &transaction_request) .await - .map_err(|e| { - TransactionQueueSendTransactionError::TransactionSendError( - SendTransactionError::InternalError(e.to_string()), - ) - })?; + .map_err(TransactionQueueSendTransactionError::TransactionSendError)?; } - let attempt_hash = Self::signed_transaction_hash(&transaction_request, signature); - - let transaction_hash = - match self.evm_provider.send_signed_transaction(transaction_request, signature).await { - Ok(hash) => hash, - Err(error) => { - // A transport-level failure is ambiguous: the node may have accepted the - // broadcast even though the response was lost ('already known' proves it - // did). Record the hash of the exact signed payload we attempted so the - // 'nonce too low' receipt check can recognise the broadcast as our own if - // it mines. A definitive node rejection means this payload is NOT in the - // mempool, so the previously recorded candidate must be kept. - if !was_previously_sent - && !Self::send_error_rules_out_broadcast(&error.to_string().to_lowercase()) - { - self.record_broadcast_attempt_hash(db, transaction, attempt_hash).await; - } - return Err(TransactionQueueSendTransactionError::TransactionSendError(error)); - } - }; - let transaction_sent = TransactionSentWithRelayer { id: transaction.id, - hash: transaction_hash, + hash: signed_transaction.hash(), sent_with_gas: gas_price, sent_with_blob_gas, }; - transaction.known_transaction_hash = Some(transaction_sent.hash); - transaction.sent_with_max_fee_per_gas = Some(transaction_sent.sent_with_gas.max_fee); - transaction.sent_with_max_priority_fee_per_gas = - Some(transaction_sent.sent_with_gas.max_priority_fee); - transaction.sent_with_gas = Some(transaction_sent.sent_with_gas.clone()); - transaction.sent_with_blob_gas = transaction_sent.sent_with_blob_gas.clone(); - transaction.sent_at = Some(Utc::now()); - transaction.status = TransactionStatus::INMEMPOOL; + persist_and_broadcast( + db, + &self.evm_provider, + &self.relayer.id, + transaction, + &transaction_sent, + &signed_transaction, + self.is_legacy_transactions(), + ) + .await?; info!( "Transaction {} sent successfully with hash {} for relayer: {}", transaction_sent.id, transaction_sent.hash, self.relayer.name ); - if !was_previously_sent || transaction.is_noop { - info!( - "Updating database for sent transaction {} on relayer: {}", - transaction.id, self.relayer.name - ); - // Persist the no-op fields before marking the transaction as sent so a crash - // between the two commits leaves a pending no-op row (safe to resend) rather - // than an inmempool row that still carries the original payload - if transaction.is_noop { - db.update_transaction_noop( - &transaction.id, - &transaction.to, - &transaction.speed, - &transaction.gas_limit.unwrap_or(GasLimit::new(21_000)), - ) - .await?; - } - - if !was_previously_sent { - db.transaction_sent( - &transaction_sent.id, - &transaction_sent.hash, - &transaction_sent.sent_with_gas, - transaction_sent.sent_with_blob_gas.as_ref(), - self.is_legacy_transactions(), - ) - .await?; - } - } else { - info!( - "Skipping DB update for gas bump transaction {} on relayer: {}", - transaction.id, self.relayer.name - ); - } - info!( "Successfully processed transaction {} for relayer: {}", transaction.id, self.relayer.name @@ -1775,6 +1735,13 @@ impl TransactionsQueue { Ok(receipt) } + pub async fn transaction_exists( + &self, + transaction_hash: &TransactionHash, + ) -> Result> { + self.evm_provider.transaction_exists(transaction_hash).await + } + pub async fn get_nonce(&self) -> Result> { let nonce = self.evm_provider.get_nonce_from_address(&self.relay_address()).await?; @@ -1796,6 +1763,7 @@ impl TransactionsQueue { let mut pending = self.pending_transactions.lock().await; if let Some(transaction) = pending.iter_mut().find(|tx| tx.id == *transaction_id) { transaction.nonce = new_nonce; + sort_pending_transactions(&mut pending); } } @@ -1817,7 +1785,277 @@ impl TransactionsQueue { #[cfg(test)] mod tests { - use super::{classify_send_error, SendErrorClass}; + use super::{ + classify_send_error, persist_and_broadcast, sort_pending_transactions, + BroadcastAttemptStore, SendErrorClass, SignedTransactionBroadcaster, + }; + use crate::{ + gas::{GasLimit, GasPriceResult, MaxFee, MaxPriorityFee}, + network::ChainId, + postgres::PostgresError, + provider::{SendTransactionError, SignedTransaction}, + relayer::RelayerId, + shared::common_types::EvmAddress, + transaction::{ + queue_system::types::{ + TransactionQueueSendTransactionError, TransactionSentWithRelayer, + }, + types::{ + Transaction, TransactionData, TransactionHash, TransactionId, TransactionNonce, + TransactionSpeed, TransactionStatus, TransactionValue, + }, + }, + }; + use alloy::primitives::TxHash; + use async_trait::async_trait; + use chrono::Utc; + use std::{ + collections::VecDeque, + sync::atomic::{AtomicUsize, Ordering}, + }; + + #[derive(Default)] + struct TestBroadcastAttemptStore { + attempts: Vec, + marks: usize, + fail_attempt: bool, + fail_mark_sent: bool, + } + + #[async_trait] + impl BroadcastAttemptStore for TestBroadcastAttemptStore { + async fn persist_attempt( + &mut self, + _relayer_id: &RelayerId, + _transaction: &Transaction, + transaction_sent: &TransactionSentWithRelayer, + _legacy_transaction: bool, + ) -> Result<(), PostgresError> { + if self.fail_attempt { + return Err(PostgresError::ConnectionPoolError(bb8::RunError::TimedOut)); + } + self.attempts.push(transaction_sent.hash); + Ok(()) + } + + async fn mark_sent( + &mut self, + _transaction_sent: &TransactionSentWithRelayer, + _legacy_transaction: bool, + ) -> Result<(), PostgresError> { + if self.fail_mark_sent { + return Err(PostgresError::ConnectionPoolError(bb8::RunError::TimedOut)); + } + self.marks += 1; + Ok(()) + } + } + + #[derive(Default)] + struct TestBroadcaster { + calls: AtomicUsize, + fail: bool, + } + + #[async_trait] + impl SignedTransactionBroadcaster for TestBroadcaster { + async fn broadcast( + &self, + transaction: &SignedTransaction, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + if self.fail { + return Err(SendTransactionError::InternalError( + "ambiguous transport failure".to_string(), + )); + } + Ok(transaction.hash()) + } + } + + fn pending_transaction() -> Transaction { + let now = Utc::now(); + Transaction { + id: TransactionId::new(), + relayer_id: RelayerId::new(), + to: EvmAddress::zero(), + from: EvmAddress::zero(), + value: TransactionValue::zero(), + data: TransactionData::empty(), + nonce: TransactionNonce::new(7), + chain_id: ChainId::new(1), + gas_limit: Some(GasLimit::new(21_000)), + status: TransactionStatus::PENDING, + blobs: None, + known_transaction_hash: None, + queued_at: now, + expires_at: now, + sent_at: None, + confirmed_at: None, + sent_with_gas: None, + sent_with_blob_gas: None, + mined_at: None, + mined_at_block_number: None, + speed: TransactionSpeed::FAST, + sent_with_max_priority_fee_per_gas: None, + sent_with_max_fee_per_gas: None, + is_noop: false, + external_id: None, + cancelled_by_transaction_id: None, + failed_reason: None, + } + } + + fn transaction_sent( + transaction: &Transaction, + hash: TransactionHash, + ) -> TransactionSentWithRelayer { + TransactionSentWithRelayer { + id: transaction.id, + hash, + sent_with_gas: GasPriceResult { + max_fee: MaxFee::new(2), + max_priority_fee: MaxPriorityFee::new(1), + min_wait_time_estimate: None, + max_wait_time_estimate: None, + }, + sent_with_blob_gas: None, + } + } + + #[test] + fn pending_transactions_remain_nonce_ordered_after_recovery_reassignment() { + let mut higher = pending_transaction(); + higher.nonce = TransactionNonce::new(9); + let mut lower = pending_transaction(); + lower.nonce = TransactionNonce::new(7); + let mut pending = VecDeque::from([higher, lower]); + + sort_pending_transactions(&mut pending); + + assert_eq!(pending[0].nonce, TransactionNonce::new(7)); + assert_eq!(pending[1].nonce, TransactionNonce::new(9)); + } + + #[tokio::test] + async fn attempt_persist_failure_prevents_raw_broadcast_and_memory_mutation() { + let mut transaction = pending_transaction(); + let hash = TransactionHash::new(TxHash::repeat_byte(3)); + let signed = SignedTransaction::for_test(hash); + let sent = transaction_sent(&transaction, hash); + let relayer_id = transaction.relayer_id; + let mut store = TestBroadcastAttemptStore { fail_attempt: true, ..Default::default() }; + let broadcaster = TestBroadcaster::default(); + + let result = persist_and_broadcast( + &mut store, + &broadcaster, + &relayer_id, + &mut transaction, + &sent, + &signed, + false, + ) + .await; + + assert!(matches!( + result, + Err(TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(_)) + )); + assert_eq!(broadcaster.calls.load(Ordering::SeqCst), 0); + assert_eq!(transaction.known_transaction_hash, None); + assert_eq!(transaction.sent_at, None); + assert_eq!(transaction.status, TransactionStatus::PENDING); + } + + #[tokio::test] + async fn ambiguous_broadcast_keeps_durable_hash_and_pending_nonce_protected() { + let mut transaction = pending_transaction(); + let hash = TransactionHash::new(TxHash::repeat_byte(4)); + let signed = SignedTransaction::for_test(hash); + let sent = transaction_sent(&transaction, hash); + let relayer_id = transaction.relayer_id; + let mut store = TestBroadcastAttemptStore::default(); + let broadcaster = TestBroadcaster { fail: true, ..Default::default() }; + + let result = persist_and_broadcast( + &mut store, + &broadcaster, + &relayer_id, + &mut transaction, + &sent, + &signed, + false, + ) + .await; + + assert!(matches!( + result, + Err(TransactionQueueSendTransactionError::TransactionSendError(_)) + )); + assert_eq!(store.attempts, vec![hash]); + assert_eq!(transaction.known_transaction_hash, Some(hash)); + assert!(transaction.sent_at.is_some()); + assert_eq!(transaction.nonce, TransactionNonce::new(7)); + assert_eq!(transaction.status, TransactionStatus::PENDING); + } + + #[tokio::test] + async fn post_broadcast_db_failure_keeps_durable_attempt_and_pending_memory_state() { + let mut transaction = pending_transaction(); + let hash = TransactionHash::new(TxHash::repeat_byte(5)); + let signed = SignedTransaction::for_test(hash); + let sent = transaction_sent(&transaction, hash); + let relayer_id = transaction.relayer_id; + let mut store = TestBroadcastAttemptStore { fail_mark_sent: true, ..Default::default() }; + let broadcaster = TestBroadcaster::default(); + + let result = persist_and_broadcast( + &mut store, + &broadcaster, + &relayer_id, + &mut transaction, + &sent, + &signed, + false, + ) + .await; + + assert!(matches!( + result, + Err(TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(_)) + )); + assert_eq!(broadcaster.calls.load(Ordering::SeqCst), 1); + assert_eq!(store.attempts, vec![hash]); + assert_eq!(transaction.known_transaction_hash, Some(hash)); + assert_eq!(transaction.status, TransactionStatus::PENDING); + } + + #[tokio::test] + async fn sent_state_advances_in_memory_only_after_database_mark_succeeds() { + let mut transaction = pending_transaction(); + let hash = TransactionHash::new(TxHash::repeat_byte(6)); + let signed = SignedTransaction::for_test(hash); + let sent = transaction_sent(&transaction, hash); + let relayer_id = transaction.relayer_id; + let mut store = TestBroadcastAttemptStore::default(); + let broadcaster = TestBroadcaster::default(); + + persist_and_broadcast( + &mut store, + &broadcaster, + &relayer_id, + &mut transaction, + &sent, + &signed, + false, + ) + .await + .unwrap(); + + assert_eq!(store.marks, 1); + assert_eq!(transaction.status, TransactionStatus::INMEMPOOL); + } #[test] fn classify_send_error_covers_node_wordings() { diff --git a/crates/core/src/transaction/queue_system/transactions_queues.rs b/crates/core/src/transaction/queue_system/transactions_queues.rs index ba7d07b1..40397bc9 100644 --- a/crates/core/src/transaction/queue_system/transactions_queues.rs +++ b/crates/core/src/transaction/queue_system/transactions_queues.rs @@ -5,9 +5,9 @@ use std::{ use alloy::{ consensus::TypedTransaction, - network::AnyTransactionReceipt, transports::{RpcError, TransportErrorKind}, }; +use async_trait::async_trait; use chrono::{DateTime, Utc}; use thiserror::Error; use tokio::sync::Mutex; @@ -23,7 +23,8 @@ pub enum TransactionsQueuesError { } use super::{ - start::spawn_processing_tasks_for_relayer, + attempts::find_landed_attempt, + start::{effective_startup_nonce, spawn_processing_tasks_for_relayer}, transactions_queue::{classify_send_error, SendErrorClass, TransactionsQueue}, types::{ AddTransactionError, CancelTransactionError, CancelTransactionResult, CompetitionType, @@ -52,10 +53,11 @@ use crate::{ shutdown::enter_critical_operation, transaction::{ cache::invalidate_transaction_no_state_cache, + db::RecordedTransactionAttempt, nonce_manager::NonceManager, queue_system::types::TransactionQueueSendTransactionError, types::{ - Transaction, TransactionData, TransactionHash, TransactionId, TransactionStatus, + Transaction, TransactionData, TransactionId, TransactionNonce, TransactionStatus, TransactionValue, }, }, @@ -65,6 +67,38 @@ use crate::{ const SAME_NONCE_BUMP_DIVISOR: u128 = 5; const MIN_SAME_NONCE_GAS_BUMP_WEI: u128 = 1_000_000_000; +#[async_trait] +trait NonceUpdateStore { + async fn persist_nonce( + &mut self, + transaction_id: &TransactionId, + nonce: &crate::transaction::types::TransactionNonce, + ) -> Result<(), crate::postgres::PostgresError>; +} + +#[async_trait] +impl NonceUpdateStore for PostgresClient { + async fn persist_nonce( + &mut self, + transaction_id: &TransactionId, + nonce: &crate::transaction::types::TransactionNonce, + ) -> Result<(), crate::postgres::PostgresError> { + self.transaction_update_nonce(transaction_id, nonce).await + } +} + +async fn persist_then_advance_transaction_nonce( + store: &mut S, + nonce_manager: &NonceManager, + transaction: &mut Transaction, + new_nonce: crate::transaction::types::TransactionNonce, +) -> Result<(), crate::postgres::PostgresError> { + store.persist_nonce(&transaction.id, &new_nonce).await?; + nonce_manager.advance_after_persisted_reservation(new_nonce).await; + transaction.nonce = new_nonce; + Ok(()) +} + fn bump_u128_for_same_nonce_competitor(value: u128) -> u128 { value .saturating_add(std::cmp::max(value / SAME_NONCE_BUMP_DIVISOR, MIN_SAME_NONCE_GAS_BUMP_WEI)) @@ -108,11 +142,19 @@ impl TransactionsQueues { let mut relayer_block_times_ms = HashMap::new(); for setup in setups { - let current_nonce = setup.evm_provider.get_nonce(&setup.relayer).await?; + let onchain_nonce = setup.evm_provider.get_nonce(&setup.relayer).await?; + let current_nonce = effective_startup_nonce( + onchain_nonce, + &setup.pending_transactions, + &setup.inmempool_transactions, + ); info!( - "Startup nonce synchronization for relayer {} ({}): synchronizing nonce manager with on-chain nonce {}", - setup.relayer.name, setup.relayer.id, current_nonce.into_inner() + "Startup nonce synchronization for relayer {} ({}): on-chain nonce {}, protected queue nonce {}", + setup.relayer.name, + setup.relayer.id, + onchain_nonce.into_inner(), + current_nonce.into_inner() ); relayer_block_times_ms.insert(setup.relayer.id, setup.evm_provider.blocks_every); @@ -245,7 +287,6 @@ impl TransactionsQueues { /// Replaces the content of an existing transaction with new parameters. fn transaction_replace( - &self, current_transaction: &mut Transaction, replace_with: &RelayTransactionRequest, ) { @@ -269,6 +310,10 @@ impl TransactionsQueues { current_transaction.external_id = replace_with.external_id.clone(); } + fn competitor_nonce(original_transaction: &Transaction) -> TransactionNonce { + original_transaction.nonce + } + /// Computes gas prices for a transaction based on its type. async fn compute_transaction_gas_prices( transactions_queue: &TransactionsQueue, @@ -470,21 +515,10 @@ impl TransactionsQueues { transaction.nonce = assigned_nonce; transaction.gas_limit = Some(estimated_gas_limit); - let transaction_request = Self::create_typed_transaction( - &transactions_queue, - &transaction, - &gas_price, - blob_gas_price.as_ref(), - estimated_gas_limit, - )?; - - transaction.known_transaction_hash = - Some(transactions_queue.compute_tx_hash(&transaction_request).await?); - - self.db - .save_transaction(relayer_id, &transaction) - .await - .map_err(AddTransactionError::CouldNotSaveTransactionDb)?; + if let Err(error) = self.db.save_transaction(relayer_id, &transaction).await { + transactions_queue.nonce_manager.release_unbroadcast_nonce(assigned_nonce).await; + return Err(AddTransactionError::CouldNotSaveTransactionDb(error)); + } transactions_queue.add_pending_transaction(transaction.clone()).await; self.invalidate_transaction_cache(&transaction.id).await; @@ -535,15 +569,15 @@ impl TransactionsQueues { result.transaction.sent_with_max_priority_fee_per_gas = None; result.transaction.sent_at = None; result.transaction.status = TransactionStatus::PENDING; - transactions_queue - .update_pending_transaction(result.transaction.clone()) - .await; - self.db .transaction_update(&result.transaction) .await .map_err(CancelTransactionError::CouldNotUpdateTransactionDb)?; + transactions_queue + .update_pending_transaction(result.transaction.clone()) + .await; + self.invalidate_transaction_cache(&transaction.id).await; if let Some(webhook_manager) = &self.webhook_manager { @@ -572,7 +606,7 @@ impl TransactionsQueues { value: TransactionValue::zero(), data: TransactionData::empty(), // Use the same nonce as the original transaction to replace it - nonce: result.transaction.nonce, + nonce: Self::competitor_nonce(&result.transaction), gas_limit: Some(GasLimit::new(21_000)), status: TransactionStatus::PENDING, blobs: None, @@ -649,7 +683,7 @@ impl TransactionsQueues { if let Err(sync_error) = self .recover_nonce_synchronization( &transaction.relayer_id, - &mut transactions_queue, + &transactions_queue, ) .await { @@ -681,8 +715,14 @@ impl TransactionsQueues { cancel_transaction.known_transaction_hash = Some(transaction_sent.hash); cancel_transaction.sent_at = Some(Utc::now()); + // Now we can safely set the foreign key reference and update the original transaction + // For now, we track that a cancellation is pending by setting the cancelled_by_transaction_id + // but keep the original status as INMEMPOOL since it's still competing + result.transaction.cancelled_by_transaction_id = + Some(cancel_transaction_id); + self.db - .save_transaction(&transaction.relayer_id, &cancel_transaction) + .transaction_update(&result.transaction) .await .map_err(CancelTransactionError::CouldNotUpdateTransactionDb)?; @@ -695,17 +735,6 @@ impl TransactionsQueues { .await .map_err(CancelTransactionError::SendTransactionError)?; - // Now we can safely set the foreign key reference and update the original transaction - // For now, we track that a cancellation is pending by setting the cancelled_by_transaction_id - // but keep the original status as INMEMPOOL since it's still competing - result.transaction.cancelled_by_transaction_id = - Some(cancel_transaction_id); - - self.db - .transaction_update(&result.transaction) - .await - .map_err(CancelTransactionError::CouldNotUpdateTransactionDb)?; - self.invalidate_transaction_cache(&transaction.id).await; info!("cancel_transaction: sent cancel tx {} with hash {} and nonce {:?} to replace original tx {}", @@ -771,7 +800,16 @@ impl TransactionsQueues { match result.type_name { EditableTransactionType::Pending => { let original_transaction = result.transaction.clone(); - self.transaction_replace(&mut result.transaction, replace_with); + Self::transaction_replace(&mut result.transaction, replace_with); + + self.db + .transaction_update(&result.transaction) + .await + .map_err(ReplaceTransactionError::CouldNotUpdateTransactionInDb)?; + transactions_queue + .update_pending_transaction(result.transaction.clone()) + .await; + self.invalidate_transaction_cache(&transaction.id).await; if let Some(webhook_manager) = &self.webhook_manager { @@ -807,7 +845,7 @@ impl TransactionsQueues { value: replace_with.value, data: replace_with.data.clone(), // Use the same nonce as the original transaction to replace it - nonce: result.transaction.nonce, + nonce: Self::competitor_nonce(&result.transaction), gas_limit: None, // Will be estimated during send_transaction status: TransactionStatus::PENDING, blobs: replace_with @@ -927,7 +965,7 @@ impl TransactionsQueues { if let Err(sync_error) = self .recover_nonce_synchronization( &transaction.relayer_id, - &mut transactions_queue, + &transactions_queue, ) .await { @@ -955,6 +993,13 @@ impl TransactionsQueues { replace_transaction.known_transaction_hash = Some(transaction_sent.hash); replace_transaction.sent_at = Some(Utc::now()); + result.transaction.cancelled_by_transaction_id = + Some(replace_transaction_id); + self.db + .transaction_update(&result.transaction) + .await + .map_err(ReplaceTransactionError::CouldNotUpdateTransactionInDb)?; + transactions_queue .add_competitor_to_inmempool_transaction( &transaction.id, @@ -964,18 +1009,6 @@ impl TransactionsQueues { .await .map_err(ReplaceTransactionError::SendTransactionError)?; - self.db - .save_transaction(&transaction.relayer_id, &replace_transaction) - .await - .map_err(ReplaceTransactionError::CouldNotUpdateTransactionInDb)?; - - result.transaction.cancelled_by_transaction_id = - Some(replace_transaction_id); - self.db - .transaction_update(&result.transaction) - .await - .map_err(ReplaceTransactionError::CouldNotUpdateTransactionInDb)?; - self.invalidate_transaction_cache(&transaction.id).await; info!("replace_transaction: added competitive replace tx {} with hash {} and nonce {:?} to replace original tx {}", @@ -1014,10 +1047,10 @@ impl TransactionsQueues { } async fn recover_nonce_synchronization( - &mut self, + &self, relayer_id: &RelayerId, - transactions_queue: &mut TransactionsQueue, - ) -> Result<(), Box> { + transactions_queue: &TransactionsQueue, + ) -> Result> { info!("Attempting nonce recovery for relayer {}", relayer_id); let current_onchain_nonce = transactions_queue @@ -1032,16 +1065,16 @@ impl TransactionsQueues { relayer_id, current_onchain_nonce.into_inner(), current_internal_nonce.into_inner() ); - transactions_queue.nonce_manager.sync_with_onchain_nonce(current_onchain_nonce).await; - - let updated_nonce = transactions_queue.nonce_manager.get_current_nonce().await; + let next_recovery_nonce = TransactionNonce::new( + current_onchain_nonce.into_inner().max(current_internal_nonce.into_inner()), + ); info!( - "Nonce recovery completed for relayer {}: updated internal nonce to {}", + "Nonce recovery evidence collected for relayer {}: next candidate nonce {} (memory remains unchanged until database persistence)", relayer_id, - updated_nonce.into_inner() + next_recovery_nonce.into_inner() ); - Ok(()) + Ok(next_recovery_nonce) } /// Closes out a pending transaction whose payload the node has permanently rejected @@ -1077,17 +1110,18 @@ impl TransactionsQueues { transaction.sent_at = None; transaction.status = TransactionStatus::PENDING; transaction.failed_reason = Some(reason.to_string()); - transactions_queue.update_pending_transaction(transaction.clone()).await; // Single atomic write: transaction_update persists the no-op payload together // with failed_reason (and the audit row), so a crash can never separate them - if let Err(db_error) = self.db.transaction_update(transaction).await { - // In-memory queue is consistent; a restart re-runs this close-out. - error!( - "close_out_pending_transaction_as_noop: failed to persist no-op payload for transaction {}: {}", - transaction.id, db_error - ); - } + self.db.transaction_update(transaction).await.map_err(|db_error| { + ProcessPendingTransactionError::SendTransactionError( + transaction.relayer_id, + transactions_queue.relay_address(), + TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(db_error), + ) + })?; + + transactions_queue.update_pending_transaction(transaction.clone()).await; self.invalidate_transaction_cache(&transaction.id).await; @@ -1122,24 +1156,35 @@ impl TransactionsQueues { relayer_address: EvmAddress, transactions_queue: &mut TransactionsQueue, transaction: &Transaction, - mined_hash: TransactionHash, - receipt: &AnyTransactionReceipt, + attempt: &RecordedTransactionAttempt, ) -> Result, ProcessPendingTransactionError> { - // Gas fields here are bookkeeping for an already-mined broadcast - take what - // the chain actually charged from the receipt instead of quoting the oracle - let effective_gas_price = receipt.effective_gas_price; let transaction_sent = TransactionSentWithRelayer { id: transaction.id, - hash: mined_hash, - sent_with_gas: GasPriceResult { - max_fee: MaxFee::new(effective_gas_price), - max_priority_fee: MaxPriorityFee::new(effective_gas_price), - min_wait_time_estimate: None, - max_wait_time_estimate: None, - }, - sent_with_blob_gas: None, + hash: attempt.hash, + sent_with_gas: attempt + .sent_with_gas + .clone() + .expect("landed recovery validates gas evidence before resolution"), + sent_with_blob_gas: attempt.sent_with_blob_gas.clone(), }; + self.db + .transaction_sent( + &transaction_sent.id, + &transaction_sent.hash, + &transaction_sent.sent_with_gas, + transaction_sent.sent_with_blob_gas.as_ref(), + transactions_queue.is_legacy_transactions(), + ) + .await + .map_err(|db_error| { + ProcessPendingTransactionError::SendTransactionError( + *relayer_id, + relayer_address, + TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(db_error), + ) + })?; + transactions_queue .move_pending_to_inmempool(transaction, &transaction_sent) .await @@ -1151,24 +1196,6 @@ impl TransactionsQueues { ) })?; - if let Err(db_error) = self - .db - .transaction_sent( - &transaction_sent.id, - &transaction_sent.hash, - &transaction_sent.sent_with_gas, - transaction_sent.sent_with_blob_gas.as_ref(), - transactions_queue.is_legacy_transactions(), - ) - .await - { - // The in-memory queue is consistent; a restart replays this resolution. - error!( - "resolve_pending_transaction_mined: failed to persist mined hash for transaction {}: {}", - transaction.id, db_error - ); - } - self.invalidate_transaction_cache(&transaction.id).await; if let Some(webhook_manager) = &self.webhook_manager { @@ -1342,16 +1369,83 @@ impl TransactionsQueues { } else if error_class == SendErrorClass::NonceConflict { warn!("process_single_pending: nonce synchronization issue detected for relayer {}: {}", relayer_id, error); - // Before reassigning a fresh nonce, check whether OUR OWN - // broadcast of this transaction is what consumed the nonce - // (a prior lost-response send that mined). Reassigning in - // that case would execute the payload a second time. - if let Some(known_hash) = transaction.known_transaction_hash { - match transactions_queue.get_receipt(&known_hash).await { - Ok(Some(receipt)) => { + // Evaluate every durable attempt in recorded order. The live + // row contains only the latest hash, so checking it alone can + // miss an earlier attempt that landed before a later retry was + // definitively rejected. + let attempts = match self + .db + .get_recorded_transaction_attempts(&transaction.id) + .await + { + Ok(attempts) if !attempts.is_empty() => attempts, + Ok(_) => { + warn!( + "process_single_pending: transaction {} has no complete attempt evidence; retrying without touching its nonce", + transaction.id + ); + return Ok( + ProcessResult::::other( + ProcessPendingStatus::SendRetrying, + Some(&100), + ), + ); + } + Err(attempt_error) => { + warn!( + "process_single_pending: could not load attempts for transaction {} ({}); retrying without touching its nonce", + transaction.id, attempt_error + ); + return Ok( + ProcessResult::::other( + ProcessPendingStatus::SendRetrying, + Some(&100), + ), + ); + } + }; + + let landed_attempt = match find_landed_attempt( + &*transactions_queue, + &attempts, + ) + .await + { + Ok(landed) => landed, + Err(attempt_error) => { + warn!( + "process_single_pending: could not prove attempt state for transaction {} ({}); retrying without touching its nonce", + transaction.id, attempt_error + ); + return Ok( + ProcessResult::::other( + ProcessPendingStatus::SendRetrying, + Some(&100), + ), + ); + } + }; + + if let Some(attempt) = landed_attempt { + let Some(sent_with_gas) = attempt.sent_with_gas.as_ref() + else { + warn!( + "process_single_pending: landed attempt {} for transaction {} has no gas evidence; retrying without touching memory", + attempt.hash, transaction.id + ); + return Ok( + ProcessResult::::other( + ProcessPendingStatus::SendRetrying, + Some(&100), + ), + ); + }; + + match transactions_queue.get_receipt(&attempt.hash).await { + Ok(Some(_receipt)) => { info!( "process_single_pending: transaction {} already mined as {} - handing over to receipt resolution instead of reassigning its nonce", - transaction.id, known_hash + transaction.id, attempt.hash ); return self .resolve_pending_transaction_mined( @@ -1359,12 +1453,80 @@ impl TransactionsQueues { relayer_address, &mut transactions_queue, &transaction, - known_hash, - &receipt, + attempt, + ) + .await; + } + Ok(None) => { + let recovered = match self + .db + .recover_landed_transaction_attempt( + relayer_id, + &transaction.id, + &attempt.hash, + sent_with_gas, + attempt.sent_with_blob_gas.as_ref(), + transactions_queue.is_legacy_transactions(), + ) + .await + { + Ok(recovered) => recovered, + Err(db_error) => { + error!( + "process_single_pending: failed to persist landed attempt recovery for transaction {}: {}", + transaction.id, db_error + ); + return Err(ProcessPendingTransactionError::SendTransactionError( + *relayer_id, + relayer_address, + TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(db_error), + )); + } + }; + + if !recovered { + warn!( + "process_single_pending: transaction {} changed before landed attempt recovery; retrying from authoritative state", + transaction.id + ); + return Ok(ProcessResult::< + ProcessPendingStatus, + >::other( + ProcessPendingStatus::SendRetrying, + Some(&100), + )); + } + + let transaction_sent = TransactionSentWithRelayer { + id: transaction.id, + hash: attempt.hash, + sent_with_gas: sent_with_gas.clone(), + sent_with_blob_gas: attempt + .sent_with_blob_gas + .clone(), + }; + transactions_queue + .move_pending_to_inmempool( + &transaction, + &transaction_sent, ) + .await + .map_err(|move_error| { + ProcessPendingTransactionError::MovePendingTransactionToInmempoolError( + *relayer_id, + relayer_address, + move_error, + ) + })?; + self.invalidate_transaction_cache(&transaction.id) .await; + return Ok( + ProcessResult::::other( + ProcessPendingStatus::NonceSynchronized, + Some(&100), + ), + ); } - Ok(None) => {} Err(receipt_error) => { // Fail closed: without the receipt we cannot rule // out that our own broadcast consumed the nonce, @@ -1384,25 +1546,40 @@ impl TransactionsQueues { } } - if let Err(sync_error) = self + let new_nonce = match self .recover_nonce_synchronization( relayer_id, - &mut transactions_queue, + &transactions_queue, ) .await { - error!("Failed to recover nonce synchronization for relayer {}: {}", relayer_id, sync_error); + Ok(new_nonce) => new_nonce, + Err(sync_error) => { + error!("Failed to recover nonce synchronization for relayer {}: {}", relayer_id, sync_error); + return Err(ProcessPendingTransactionError::SendTransactionError( + *relayer_id, + relayer_address, + TransactionQueueSendTransactionError::TransactionSendError(error), + )); + } + }; + + if let Err(db_error) = persist_then_advance_transaction_nonce( + &mut self.db, + &transactions_queue.nonce_manager, + &mut transaction, + new_nonce, + ) + .await + { + error!("Failed to persist nonce update to database for transaction {}: {}", transaction.id, db_error); return Err(ProcessPendingTransactionError::SendTransactionError( *relayer_id, relayer_address, - TransactionQueueSendTransactionError::TransactionSendError(error), + TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(db_error), )); } - let new_nonce = - transactions_queue.nonce_manager.get_and_increment().await; - transaction.nonce = new_nonce; - transactions_queue .update_pending_transaction_nonce( &transaction.id, @@ -1410,14 +1587,6 @@ impl TransactionsQueues { ) .await; - if let Err(db_error) = self - .db - .transaction_update_nonce(&transaction.id, &new_nonce) - .await - { - error!("Failed to persist nonce update to database for transaction {}: {}", transaction.id, db_error); - } - info!("Nonce synchronization recovered for relayer {}, updated pending transaction nonce to {} in queue and database", relayer_id, new_nonce.into_inner()); Ok(ProcessResult::::other( @@ -1727,24 +1896,34 @@ impl TransactionsQueues { } } - if let Err(sync_error) = self.recover_nonce_synchronization(relayer_id, &mut transactions_queue).await { - error!("Failed to recover nonce synchronization for relayer {}: {}", relayer_id, sync_error); + let new_nonce = match self.recover_nonce_synchronization(relayer_id, &transactions_queue).await { + Ok(new_nonce) => new_nonce, + Err(sync_error) => { + error!("Failed to recover nonce synchronization for relayer {}: {}", relayer_id, sync_error); + return Err(ProcessInmempoolTransactionError::SendTransactionError( + *relayer_id, + relayer_address, + TransactionQueueSendTransactionError::TransactionSendError(error) + )); + } + }; + + if let Err(db_error) = persist_then_advance_transaction_nonce( + &mut self.db, + &transactions_queue.nonce_manager, + &mut transaction, + new_nonce, + ).await { + error!("Failed to persist nonce update to database for transaction {}: {}", transaction.id, db_error); return Err(ProcessInmempoolTransactionError::SendTransactionError( *relayer_id, relayer_address, - TransactionQueueSendTransactionError::TransactionSendError(error) + TransactionQueueSendTransactionError::CouldNotUpdateTransactionDb(db_error) )); } - let new_nonce = transactions_queue.nonce_manager.get_and_increment().await; - transaction.nonce = new_nonce; - transactions_queue.update_inmempool_transaction_nonce(&transaction.id, new_nonce).await; - if let Err(db_error) = self.db.transaction_update_nonce(&transaction.id, &new_nonce).await { - error!("Failed to persist nonce update to database for transaction {}: {}", transaction.id, db_error); - } - info!("Nonce synchronization recovered for relayer {}, updated gas bump transaction nonce {} in queue and database", relayer_id, new_nonce.into_inner()); return Ok(ProcessResult::::other( @@ -1778,31 +1957,6 @@ impl TransactionsQueues { .await; } - // Update the local transaction with the new gas values so subsequent bumps work correctly - transaction.known_transaction_hash = - Some(transaction_sent.hash); - transaction.sent_with_max_fee_per_gas = - Some(transaction_sent.sent_with_gas.max_fee); - transaction.sent_with_max_priority_fee_per_gas = - Some(transaction_sent.sent_with_gas.max_priority_fee); - transaction.sent_with_gas = - Some(transaction_sent.sent_with_gas.clone()); - transaction.sent_at = Some(Utc::now()); - - if let Err(db_error) = self - .db - .transaction_sent( - &transaction_sent.id, - &transaction_sent.hash, - &transaction_sent.sent_with_gas, - transaction_sent.sent_with_blob_gas.as_ref(), - transactions_queue.is_legacy_transactions(), - ) - .await - { - error!("Failed to persist gas bump to database for transaction {}: {}", transaction.id, db_error); - } - self.invalidate_transaction_cache(&transaction.id).await; return Ok(ProcessResult::::other( @@ -1961,3 +2115,134 @@ impl TransactionsQueues { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + gas::GasLimit, + network::ChainId, + postgres::PostgresError, + shared::common_types::EvmAddress, + transaction::types::{ + TransactionData, TransactionNonce, TransactionStatus, TransactionValue, + }, + }; + + struct TestNonceStore { + fail: bool, + } + + #[async_trait] + impl NonceUpdateStore for TestNonceStore { + async fn persist_nonce( + &mut self, + _transaction_id: &TransactionId, + _nonce: &TransactionNonce, + ) -> Result<(), PostgresError> { + if self.fail { + Err(PostgresError::ConnectionPoolError(bb8::RunError::TimedOut)) + } else { + Ok(()) + } + } + } + + fn transaction_with_nonce(nonce: u64) -> Transaction { + let now = Utc::now(); + Transaction { + id: TransactionId::new(), + relayer_id: RelayerId::new(), + to: EvmAddress::zero(), + from: EvmAddress::zero(), + value: TransactionValue::zero(), + data: TransactionData::empty(), + nonce: TransactionNonce::new(nonce), + chain_id: ChainId::new(1), + gas_limit: Some(GasLimit::new(21_000)), + status: TransactionStatus::PENDING, + blobs: None, + known_transaction_hash: None, + queued_at: now, + expires_at: now, + sent_at: None, + confirmed_at: None, + sent_with_gas: None, + sent_with_blob_gas: None, + mined_at: None, + mined_at_block_number: None, + speed: TransactionSpeed::FAST, + sent_with_max_priority_fee_per_gas: None, + sent_with_max_fee_per_gas: None, + is_noop: false, + external_id: None, + cancelled_by_transaction_id: None, + failed_reason: None, + } + } + + #[tokio::test] + async fn nonce_persistence_failure_leaves_transaction_and_manager_memory_unchanged() { + let manager = NonceManager::new(TransactionNonce::new(7)); + let mut transaction = transaction_with_nonce(3); + let mut store = TestNonceStore { fail: true }; + + let result = persist_then_advance_transaction_nonce( + &mut store, + &manager, + &mut transaction, + TransactionNonce::new(9), + ) + .await; + + assert!(result.is_err()); + assert_eq!(transaction.nonce, TransactionNonce::new(3)); + assert_eq!(manager.get_current_nonce().await, TransactionNonce::new(7)); + } + + #[tokio::test] + async fn persisted_nonce_advances_transaction_then_manager_memory() { + let manager = NonceManager::new(TransactionNonce::new(7)); + let mut transaction = transaction_with_nonce(3); + let mut store = TestNonceStore { fail: false }; + + persist_then_advance_transaction_nonce( + &mut store, + &manager, + &mut transaction, + TransactionNonce::new(9), + ) + .await + .unwrap(); + + assert_eq!(transaction.nonce, TransactionNonce::new(9)); + assert_eq!(manager.get_current_nonce().await, TransactionNonce::new(10)); + } + + #[test] + fn pending_replacement_preserves_transaction_identity_and_original_nonce() { + let mut transaction = transaction_with_nonce(7); + let original_id = transaction.id; + let replacement = RelayTransactionRequest { + to: EvmAddress::zero(), + value: TransactionValue::new(alloy::primitives::U256::from(42)), + data: TransactionData::empty(), + speed: Some(TransactionSpeed::SUPER), + external_id: Some("replacement".to_string()), + blobs: None, + }; + + TransactionsQueues::transaction_replace(&mut transaction, &replacement); + + assert_eq!(transaction.id, original_id); + assert_eq!(transaction.nonce, TransactionNonce::new(7)); + assert_eq!(transaction.external_id.as_deref(), Some("replacement")); + } + + #[test] + fn cancellation_and_replacement_competitors_use_original_nonce() { + let original = transaction_with_nonce(11); + + assert_eq!(TransactionsQueues::competitor_nonce(&original), TransactionNonce::new(11)); + } +} diff --git a/crates/core/src/wallet/composite_wallet_manager.rs b/crates/core/src/wallet/composite_wallet_manager.rs index 26c953b5..553bd697 100644 --- a/crates/core/src/wallet/composite_wallet_manager.rs +++ b/crates/core/src/wallet/composite_wallet_manager.rs @@ -4,7 +4,7 @@ use alloy::dyn_abi::TypedData; use alloy::{consensus::TypedTransaction, signers::Signature}; use async_trait::async_trait; -use crate::wallet::WalletManagerChainId; +use crate::{relayer::WalletIndex, wallet::WalletManagerChainId}; use crate::{ shared::common_types::EvmAddress, wallet::{WalletError, WalletManagerTrait}, @@ -27,7 +27,7 @@ impl CompositeWalletManager { // TODO: not ideal route but only way i could find for now to work without a big refactor /// Determine if a wallet index is for a private key (high range) fn is_private_key_index(&self, wallet_index: u32) -> bool { - wallet_index >= u32::MAX - 1000 + WalletIndex::is_private_key_manager_index(wallet_index) } /// Get the appropriate wallet manager for the given index diff --git a/crates/core/src/wallet/private_key_wallet_manager.rs b/crates/core/src/wallet/private_key_wallet_manager.rs index c0f3c07a..eac8bfa4 100644 --- a/crates/core/src/wallet/private_key_wallet_manager.rs +++ b/crates/core/src/wallet/private_key_wallet_manager.rs @@ -1,5 +1,6 @@ use crate::common_types::EvmAddress; use crate::network::ChainId; +use crate::relayer::WalletIndex; use crate::wallet::{WalletError, WalletManagerChainId, WalletManagerTrait}; use alloy::consensus::TypedTransaction; use alloy::dyn_abi::TypedData; @@ -23,7 +24,7 @@ impl PrivateKeyWalletManager { /// Convert wallet index from WalletIndex system back to internal array index fn convert_to_internal_index(&self, wallet_index: u32) -> Result { // Check if this is a private key index (high range: >= u32::MAX - 1000) - if wallet_index >= u32::MAX - 1000 { + if WalletIndex::is_private_key_manager_index(wallet_index) { // This is a private key index from the WalletIndex conversion, convert back to array index let internal_index = (u32::MAX - wallet_index) as usize; if internal_index < self.private_keys.len() { diff --git a/crates/e2e-tests/src/tests/transactions/status/confirmed.rs b/crates/e2e-tests/src/tests/transactions/status/confirmed.rs index c0a621f9..286eb1b9 100644 --- a/crates/e2e-tests/src/tests/transactions/status/confirmed.rs +++ b/crates/e2e-tests/src/tests/transactions/status/confirmed.rs @@ -61,10 +61,12 @@ impl TestRunner { let hash = status.hash.unwrap(); info!("Transaction hash: {:?}", hash); info!("Expected hash: {:?}", send_result.hash); - if hash != send_result.hash { - return Err(anyhow::anyhow!( - "Confirmed transaction should match the sent transaction hash" - )); + if let Some(expected_hash) = send_result.hash { + if hash != expected_hash { + return Err(anyhow::anyhow!( + "Confirmed transaction should match the sent transaction hash" + )); + } } if status.receipt.is_none() { return Err(anyhow::anyhow!("Confirmed transaction should have receipt")); diff --git a/crates/e2e-tests/src/tests/transactions/status/inmempool.rs b/crates/e2e-tests/src/tests/transactions/status/inmempool.rs index cba4969b..594ae88d 100644 --- a/crates/e2e-tests/src/tests/transactions/status/inmempool.rs +++ b/crates/e2e-tests/src/tests/transactions/status/inmempool.rs @@ -46,10 +46,12 @@ impl TestRunner { let hash = status.hash.unwrap(); info!("Transaction hash: {:?}", hash); info!("Expected hash: {:?}", send_result.hash); - if hash != send_result.hash { - return Err(anyhow::anyhow!( - "InMempool transaction should match the sent transaction hash" - )); + if let Some(expected_hash) = send_result.hash { + if hash != expected_hash { + return Err(anyhow::anyhow!( + "InMempool transaction should match the sent transaction hash" + )); + } } if status.receipt.is_some() { diff --git a/crates/e2e-tests/src/tests/transactions/status/mined.rs b/crates/e2e-tests/src/tests/transactions/status/mined.rs index b136623a..662e1205 100644 --- a/crates/e2e-tests/src/tests/transactions/status/mined.rs +++ b/crates/e2e-tests/src/tests/transactions/status/mined.rs @@ -62,10 +62,12 @@ impl TestRunner { let hash = status.hash.unwrap(); info!("Transaction hash: {:?}", hash); info!("Expected hash: {:?}", send_result.hash); - if hash != send_result.hash { - return Err(anyhow::anyhow!( - "Mined transaction should match the sent transaction hash" - )); + if let Some(expected_hash) = send_result.hash { + if hash != expected_hash { + return Err(anyhow::anyhow!( + "Mined transaction should match the sent transaction hash" + )); + } } if status.receipt.is_none() { diff --git a/documentation/rrelayer/docs/pages/changelog.mdx b/documentation/rrelayer/docs/pages/changelog.mdx index 87b3c498..d3892503 100644 --- a/documentation/rrelayer/docs/pages/changelog.mdx +++ b/documentation/rrelayer/docs/pages/changelog.mdx @@ -10,10 +10,14 @@ ### Bug fixes +- fix: persist signed broadcast evidence before rpc send and recover ambiguous attempts without reusing protected nonces + --- ### Breaking changes +- transaction submission responses keep `hash` present but allow `null` after durable acceptance; poll by `id` + --- ## Releases diff --git a/documentation/rrelayer/docs/pages/integration/sdk/transactions/node.mdx b/documentation/rrelayer/docs/pages/integration/sdk/transactions/node.mdx index c7631596..6f725002 100644 --- a/documentation/rrelayer/docs/pages/integration/sdk/transactions/node.mdx +++ b/documentation/rrelayer/docs/pages/integration/sdk/transactions/node.mdx @@ -47,10 +47,13 @@ Just so you understand the properties, the properties you get back when you send ```ts export interface TransactionSent { id: string; - hash: `0x${string}`; + hash: `0x${string}` | null; } ``` +`null` is a successful, durably accepted submission whose broadcast hash is not available yet. +Poll the transaction by `id`; do not resubmit it. + ### Sending Native Transaction You can send native funds using the below: diff --git a/documentation/rrelayer/docs/pages/integration/sdk/transactions/rust.mdx b/documentation/rrelayer/docs/pages/integration/sdk/transactions/rust.mdx index 717f00ca..d1ea84e2 100644 --- a/documentation/rrelayer/docs/pages/integration/sdk/transactions/rust.mdx +++ b/documentation/rrelayer/docs/pages/integration/sdk/transactions/rust.mdx @@ -73,10 +73,13 @@ Just so you understand the properties, the properties you get back when you send #[derive(Debug, Serialize, Deserialize)] pub struct SendTransactionResult { pub id: TransactionId, - pub hash: TransactionHash, + pub hash: Option, } ``` +`None` is a successful, durably accepted submission whose broadcast hash is not available yet. +Poll the transaction by `id`; do not resubmit it. + ### Sending Native Transaction You can send native funds using the below: diff --git a/sdk/typescript/src/api/transaction/types.contract.d.ts b/sdk/typescript/src/api/transaction/types.contract.d.ts new file mode 100644 index 00000000..d6dd5093 --- /dev/null +++ b/sdk/typescript/src/api/transaction/types.contract.d.ts @@ -0,0 +1,17 @@ +import type { TransactionSent } from './types'; + +type Assert = T; +type Equal = + (() => Value extends Left ? 1 : 2) extends < + Value, + >() => Value extends Right ? 1 : 2 + ? true + : false; + +export type PendingTransactionSentContract = Assert< + { id: string; hash: null } extends TransactionSent ? true : false +>; + +export type TransactionSentHashContract = Assert< + Equal +>; diff --git a/sdk/typescript/src/api/transaction/types.ts b/sdk/typescript/src/api/transaction/types.ts index 1b10e627..f7c01ef8 100644 --- a/sdk/typescript/src/api/transaction/types.ts +++ b/sdk/typescript/src/api/transaction/types.ts @@ -65,5 +65,6 @@ export interface TransactionToSend { export interface TransactionSent { id: string; - hash: `0x${string}`; + /** Null means the transaction was durably accepted and is awaiting broadcast. */ + hash: `0x${string}` | null; }