use std::num::NonZeroUsize;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use alloy_network::{Ethereum as AlloyEthereum, EthereumWallet, NetworkWallet, TransactionBuilder};
use alloy_primitives::{Address, Bytes};
use alloy_provider::fillers::{
BlobGasFiller, ChainIdFiller, FillProvider, GasFiller, JoinFill, NonceFiller, WalletFiller,
};
use alloy_provider::{
Identity, PendingTransactionError, Provider, ProviderBuilder, RootProvider, WalletProvider,
};
use alloy_rpc_client::RpcClient;
use alloy_rpc_types_eth::{BlockId, TransactionReceipt, TransactionRequest};
use alloy_transport::TransportError;
use alloy_transport::layers::{FallbackLayer, ThrottleLayer};
use alloy_transport_http::Http;
use r402::chain::{ChainId, ChainProvider};
use tower::ServiceBuilder;
#[cfg(feature = "telemetry")]
use tracing::Instrument;
use url::Url;
use crate::chain::nonce::PendingNonceManager;
use crate::chain::types::Eip155ChainReference;
pub type InnerFiller = JoinFill<
GasFiller,
JoinFill<BlobGasFiller, JoinFill<NonceFiller<PendingNonceManager>, ChainIdFiller>>,
>;
pub type InnerProvider = FillProvider<
JoinFill<JoinFill<Identity, InnerFiller>, WalletFiller<EthereumWallet>>,
RootProvider,
>;
#[derive(Debug)]
pub struct Eip155ChainProvider {
chain: Eip155ChainReference,
eip1559: bool,
flashblocks: bool,
receipt_timeout_secs: u64,
inner: InnerProvider,
signer_addresses: Arc<Vec<Address>>,
signer_cursor: Arc<AtomicUsize>,
nonce_manager: PendingNonceManager,
}
impl Eip155ChainProvider {
pub fn new(
chain: Eip155ChainReference,
wallet: EthereumWallet,
rpc_endpoints: &[(Url, Option<u32>)],
eip1559: bool,
flashblocks: bool,
receipt_timeout_secs: u64,
) -> Result<Self, Box<dyn std::error::Error>> {
let signer_addresses =
NetworkWallet::<AlloyEthereum>::signer_addresses(&wallet).collect::<Vec<_>>();
if signer_addresses.is_empty() {
return Err("at least one signer must be provided".into());
}
let signer_addresses = Arc::new(signer_addresses);
let signer_cursor = Arc::new(AtomicUsize::new(0));
let chain_id: ChainId = chain.into();
let client = Self::rpc_client(&chain_id, rpc_endpoints);
let nonce_manager = PendingNonceManager::default();
let filler = JoinFill::new(
GasFiller,
JoinFill::new(
BlobGasFiller::default(),
JoinFill::new(
NonceFiller::new(nonce_manager.clone()),
ChainIdFiller::default(),
),
),
);
let inner: InnerProvider = ProviderBuilder::default()
.filler(filler)
.wallet(wallet)
.connect_client(client);
#[cfg(feature = "telemetry")]
tracing::info!(chain=%chain_id, signers=?signer_addresses, "Using EVM provider");
Ok(Self {
chain,
eip1559,
flashblocks,
receipt_timeout_secs,
inner,
signer_addresses,
signer_cursor,
nonce_manager,
})
}
#[must_use]
pub fn rpc_client(chain_id: &ChainId, endpoints: &[(Url, Option<u32>)]) -> RpcClient {
#[cfg(not(feature = "telemetry"))]
let _ = chain_id;
let transports = endpoints
.iter()
.filter_map(|(url, rate_limit)| {
let scheme = url.scheme();
let is_http = scheme == "http" || scheme == "https";
if !is_http {
return None;
}
#[cfg(feature = "telemetry")]
tracing::info!(chain=%chain_id, rpc_url=%url, rate_limit=?rate_limit, "Using HTTP transport");
let limit = rate_limit.unwrap_or(u32::MAX);
let service = ServiceBuilder::new()
.layer(ThrottleLayer::new(limit))
.service(Http::new(url.clone()));
Some(service)
})
.collect::<Vec<_>>();
let fallback = ServiceBuilder::new()
.layer(
FallbackLayer::default().with_active_transport_count(
NonZeroUsize::new(transports.len())
.expect("Non-zero amount of stateless transports"),
),
)
.service(transports);
RpcClient::new(fallback, false)
}
fn next_signer_address(&self) -> Address {
debug_assert!(!self.signer_addresses.is_empty());
if self.signer_addresses.len() == 1 {
self.signer_addresses[0]
} else {
let next =
self.signer_cursor.fetch_add(1, Ordering::Relaxed) % self.signer_addresses.len();
self.signer_addresses[next]
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum MetaTransactionSendError {
#[error(transparent)]
Transport(#[from] TransportError),
#[error(transparent)]
PendingTransaction(#[from] PendingTransactionError),
#[error("{0}")]
Custom(String),
}
#[derive(Debug)]
pub struct MetaTransaction {
pub to: Address,
pub calldata: Bytes,
pub confirmations: u64,
}
impl ChainProvider for Eip155ChainProvider {
fn signer_addresses(&self) -> Vec<String> {
self.inner
.signer_addresses()
.map(|a| a.to_string())
.collect()
}
fn chain_id(&self) -> ChainId {
self.chain.into()
}
}
pub trait Eip155MetaTransactionProvider {
type Error;
type Inner: Provider;
fn inner(&self) -> &Self::Inner;
fn chain(&self) -> &Eip155ChainReference;
fn send_transaction(
&self,
tx: MetaTransaction,
) -> impl Future<Output = Result<TransactionReceipt, Self::Error>> + Send;
}
impl<T: Eip155MetaTransactionProvider> Eip155MetaTransactionProvider for Arc<T> {
type Error = T::Error;
type Inner = T::Inner;
fn inner(&self) -> &Self::Inner {
(**self).inner()
}
fn chain(&self) -> &Eip155ChainReference {
(**self).chain()
}
fn send_transaction(
&self,
tx: MetaTransaction,
) -> impl Future<Output = Result<TransactionReceipt, Self::Error>> + Send {
(**self).send_transaction(tx)
}
}
impl Eip155MetaTransactionProvider for Eip155ChainProvider {
type Error = MetaTransactionSendError;
type Inner = InnerProvider;
fn inner(&self) -> &Self::Inner {
&self.inner
}
fn chain(&self) -> &Eip155ChainReference {
&self.chain
}
async fn send_transaction(
&self,
tx: MetaTransaction,
) -> Result<TransactionReceipt, Self::Error> {
let from_address = self.next_signer_address();
let mut txr = TransactionRequest::default()
.with_to(tx.to)
.with_from(from_address)
.with_input(tx.calldata);
if !self.eip1559 {
let provider = &self.inner;
let gas_fut = provider.get_gas_price();
#[cfg(feature = "telemetry")]
let gas: u128 = gas_fut
.instrument(tracing::info_span!("get_gas_price"))
.await?;
#[cfg(not(feature = "telemetry"))]
let gas: u128 = gas_fut.await?;
txr.set_gas_price(gas);
}
if txr.gas.is_none() {
let block_id = if self.flashblocks {
BlockId::latest()
} else {
BlockId::pending()
};
let gas_limit = self.inner.estimate_gas(txr.clone()).block(block_id).await?;
txr.set_gas_limit(gas_limit);
}
let pending_tx = match self.inner.send_transaction(txr).await {
Ok(pending) => pending,
Err(e) => {
self.nonce_manager.reset_nonce(from_address).await;
return Err(MetaTransactionSendError::Transport(e));
}
};
let timeout = std::time::Duration::from_secs(self.receipt_timeout_secs);
let watcher = pending_tx
.with_required_confirmations(tx.confirmations)
.with_timeout(Some(timeout));
match watcher.get_receipt().await {
Ok(receipt) => Ok(receipt),
Err(e) => {
self.nonce_manager.reset_nonce(from_address).await;
Err(MetaTransactionSendError::PendingTransaction(e))
}
}
}
}