1use super::{EthApiSpec, EthSigner, LoadBlock, LoadFee, LoadReceipt, LoadState, SpawnBlocking};
5use crate::{
6 helpers::{estimate::EstimateCall, spec::SignersForRpc},
7 FromEthApiError, FullEthApiTypes, IntoEthApiError, RpcNodeCore, RpcNodeCoreExt, RpcReceipt,
8 RpcTransaction,
9};
10use alloy_consensus::{
11 transaction::{SignerRecoverable, TransactionMeta, TxHashRef},
12 BlockHeader, Transaction,
13};
14use alloy_dyn_abi::TypedData;
15use alloy_eips::{eip2718::Encodable2718, BlockId};
16use alloy_network::{TransactionBuilder, TransactionBuilder4844};
17use alloy_primitives::{Address, Bytes, TxHash, B256, U256};
18use alloy_rpc_types_eth::{state::EvmOverrides, TransactionInfo};
19use futures::{Future, StreamExt};
20use reth_chain_state::CanonStateSubscriptions;
21use reth_primitives_traits::{
22 BlockBody, Recovered, RecoveredBlock, SignedTransaction, TxTy, WithEncoded,
23};
24use reth_rpc_convert::{transaction::RpcConvert, RpcTxReq, TransactionConversionError};
25use reth_rpc_eth_types::{
26 block::convert_transaction_receipt,
27 utils::binary_search,
28 EthApiError::{self, TransactionConfirmationTimeout},
29 FillTransaction, SignError, TransactionSource,
30};
31use reth_storage_api::{
32 BlockNumReader, BlockReaderIdExt, ProviderBlock, ProviderReceipt, ProviderTx, ReceiptProvider,
33 TransactionsProvider,
34};
35use reth_transaction_pool::{
36 AddedTransactionOutcome, PoolPooledTx, PoolTransaction, PoolTx, TransactionOrigin,
37 TransactionPool,
38};
39use std::{sync::Arc, time::Duration};
40
41pub trait EthTransactions: LoadTransaction<Provider: BlockReaderIdExt> {
64 fn signers(&self) -> &SignersForRpc<Self::Provider, Self::NetworkTypes>;
68
69 fn accounts(&self) -> Vec<Address> {
71 self.signers().read().iter().flat_map(|s| s.accounts()).collect()
72 }
73
74 fn send_raw_transaction_sync_timeout(&self) -> Duration;
76
77 fn send_raw_transaction(
81 &self,
82 tx: Bytes,
83 ) -> impl Future<Output = Result<B256, Self::Error>> + Send {
84 async move {
85 let pool_transaction =
86 <PoolTx<Self::Pool> as PoolTransaction>::recover_raw_transaction(&tx)
87 .map_err(Self::Error::from_eth_err)?;
88 self.send_pool_transaction(
89 TransactionOrigin::Local,
90 WithEncoded::new(tx, pool_transaction),
91 )
92 .await
93 }
94 }
95
96 fn send_transaction(
98 &self,
99 origin: TransactionOrigin,
100 tx: WithEncoded<Recovered<PoolPooledTx<Self::Pool>>>,
101 ) -> impl Future<Output = Result<B256, Self::Error>> + Send {
102 async move {
103 let (encoded, recovered) = tx.split();
104 let pool_transaction =
105 <Self::Pool as TransactionPool>::Transaction::from_pooled(recovered);
106
107 self.send_pool_transaction(origin, WithEncoded::new(encoded, pool_transaction)).await
108 }
109 }
110
111 fn send_pool_transaction(
113 &self,
114 origin: TransactionOrigin,
115 tx: WithEncoded<PoolTx<Self::Pool>>,
116 ) -> impl Future<Output = Result<B256, Self::Error>> + Send;
117
118 fn send_raw_transaction_sync(
122 &self,
123 tx: Bytes,
124 timeout_ms: Option<u64>,
125 ) -> impl Future<Output = Result<RpcReceipt<Self::NetworkTypes>, Self::Error>> + Send
126 where
127 Self: LoadReceipt + 'static,
128 {
129 let this = self.clone();
130 let configured_timeout = self.send_raw_transaction_sync_timeout();
131 let timeout_duration = timeout_ms
132 .filter(|timeout_ms| *timeout_ms > 0)
133 .map(Duration::from_millis)
134 .map(|timeout| timeout.min(configured_timeout))
135 .unwrap_or(configured_timeout);
136 async move {
137 let mut stream = this.provider().canonical_state_stream();
138 let hash = EthTransactions::send_raw_transaction(&this, tx).await?;
139 tokio::time::timeout(timeout_duration, async {
140 while let Some(notification) = stream.next().await {
141 let chain = notification.committed();
142 if let Some((block, tx, receipt, all_receipts)) =
143 chain.find_transaction_and_receipt_by_hash(hash) &&
144 let Some(receipt) = convert_transaction_receipt(
145 block,
146 all_receipts,
147 tx,
148 receipt,
149 this.converter(),
150 )
151 .transpose()
152 .map_err(Self::Error::from)?
153 {
154 return Ok(receipt);
155 }
156 }
157 Err(Self::Error::from_eth_err(TransactionConfirmationTimeout {
158 hash,
159 duration: timeout_duration,
160 }))
161 })
162 .await
163 .unwrap_or_else(|_elapsed| {
164 Err(Self::Error::from_eth_err(TransactionConfirmationTimeout {
165 hash,
166 duration: timeout_duration,
167 }))
168 })
169 }
170 }
171
172 #[expect(clippy::complexity)]
178 fn transaction_by_hash(
179 &self,
180 hash: B256,
181 ) -> impl Future<
182 Output = Result<Option<TransactionSource<ProviderTx<Self::Provider>>>, Self::Error>,
183 > + Send {
184 LoadTransaction::transaction_by_hash(self, hash)
185 }
186
187 fn pending_transactions(&self) -> Result<Vec<RpcTransaction<Self::NetworkTypes>>, Self::Error> {
189 self.pool()
190 .pending_transactions()
191 .into_iter()
192 .map(|tx| self.converter().fill_pending(tx.transaction.clone_into_consensus()))
193 .collect::<Result<Vec<_>, _>>()
194 .map_err(Self::Error::from)
195 }
196
197 #[expect(clippy::type_complexity)]
201 fn transactions_by_block(
202 &self,
203 block: B256,
204 ) -> impl Future<Output = Result<Option<Vec<ProviderTx<Self::Provider>>>, Self::Error>> + Send
205 {
206 async move {
207 self.cache()
208 .get_recovered_block(block)
209 .await
210 .map(|b| b.map(|b| b.body().transactions().to_vec()))
211 .map_err(Self::Error::from_eth_err)
212 }
213 }
214
215 fn raw_transaction_by_hash(
223 &self,
224 hash: B256,
225 ) -> impl Future<Output = Result<Option<Bytes>, Self::Error>> + Send {
226 async move {
227 if let Some(tx) =
229 self.pool().get_pooled_transaction_element(hash).map(|tx| tx.encoded_2718().into())
230 {
231 return Ok(Some(tx))
232 }
233
234 self.spawn_blocking_io(move |ref this| {
235 Ok(this
236 .provider()
237 .transaction_by_hash(hash)
238 .map_err(Self::Error::from_eth_err)?
239 .map(|tx| tx.encoded_2718().into()))
240 })
241 .await
242 }
243 }
244
245 #[expect(clippy::type_complexity)]
247 fn historical_transaction_by_hash_at(
248 &self,
249 hash: B256,
250 ) -> impl Future<
251 Output = Result<Option<(TransactionSource<ProviderTx<Self::Provider>>, B256)>, Self::Error>,
252 > + Send {
253 async move {
254 match self.transaction_by_hash_at(hash).await? {
255 None => Ok(None),
256 Some((tx, at)) => Ok(at.as_block_hash().map(|hash| (tx, hash))),
257 }
258 }
259 }
260
261 fn transaction_receipt(
266 &self,
267 hash: B256,
268 ) -> impl Future<Output = Result<Option<RpcReceipt<Self::NetworkTypes>>, Self::Error>> + Send
269 where
270 Self: LoadReceipt + 'static,
271 {
272 async move {
273 match self.load_transaction_and_receipt(hash).await? {
274 Some((tx, meta, receipt, all_receipts, block)) => self
275 .build_transaction_receipt(tx, meta, receipt, all_receipts, block)
276 .await
277 .map(Some),
278 None => Ok(None),
279 }
280 }
281 }
282
283 #[expect(clippy::complexity)]
287 fn load_transaction_and_receipt(
288 &self,
289 hash: TxHash,
290 ) -> impl Future<
291 Output = Result<
292 Option<(
293 Recovered<ProviderTx<Self::Provider>>,
294 TransactionMeta,
295 ProviderReceipt<Self::Provider>,
296 Option<Arc<Vec<ProviderReceipt<Self::Provider>>>>,
297 Option<Arc<RecoveredBlock<ProviderBlock<Self::Provider>>>>,
298 )>,
299 Self::Error,
300 >,
301 > + Send
302 where
303 Self: 'static,
304 {
305 async move {
306 if let Some(cached) = self.cache().get_transaction_by_hash(hash).await &&
307 let Some(tx) = cached.recovered_transaction().map(|tx| tx.cloned())
308 {
309 let meta = cached.transaction_meta(hash);
310
311 if let Some(all_receipts) = cached.receipts.clone() &&
313 let Some(receipt) = all_receipts.get(cached.tx_index).cloned()
314 {
315 return Ok(Some((tx, meta, receipt, Some(all_receipts), Some(cached.block))));
316 }
317
318 if let Some(receipts) = self
322 .cache()
323 .get_receipts(cached.block.hash())
324 .await
325 .map_err(Self::Error::from_eth_err)? &&
326 let Some(receipt) = receipts.get(cached.tx_index).cloned()
327 {
328 return Ok(Some((tx, meta, receipt, Some(receipts), Some(cached.block))));
329 }
330 }
331
332 self.spawn_blocking_io(move |this| {
334 let provider = this.provider();
335 let Some((tx, meta)) = provider
336 .transaction_by_hash_with_meta(hash)
337 .map_err(Self::Error::from_eth_err)?
338 else {
339 return Ok(None);
340 };
341
342 let tx = tx.try_into_recovered_unchecked().map_err(Self::Error::from_eth_err)?;
343
344 let receipt = provider.receipt_by_hash(hash).map_err(Self::Error::from_eth_err)?;
345
346 Ok(receipt.map(|receipt| (tx, meta, receipt, None, None)))
347 })
348 .await
349 }
350 }
351
352 fn transaction_by_block_and_tx_index(
356 &self,
357 block_id: BlockId,
358 index: usize,
359 ) -> impl Future<Output = Result<Option<RpcTransaction<Self::NetworkTypes>>, Self::Error>> + Send
360 where
361 Self: LoadBlock,
362 {
363 async move {
364 if let Some(block) = self.recovered_block(block_id).await? {
365 let block_hash = block.hash();
366 let block_number = block.number();
367 let block_timestamp = block.timestamp();
368 let base_fee_per_gas = block.base_fee_per_gas();
369 if let Some((signer, tx)) = block.transactions_with_sender().nth(index) {
370 let tx_info = TransactionInfo {
371 hash: Some(*tx.tx_hash()),
372 block_hash: Some(block_hash),
373 block_number: Some(block_number),
374 block_timestamp: Some(block_timestamp),
375 base_fee: base_fee_per_gas,
376 index: Some(index as u64),
377 };
378
379 return Ok(Some(
380 self.converter().fill(tx.clone().with_signer(*signer), tx_info)?,
381 ))
382 }
383 }
384
385 Ok(None)
386 }
387 }
388
389 fn get_transaction_by_sender_and_nonce(
391 &self,
392 sender: Address,
393 nonce: u64,
394 include_pending: bool,
395 ) -> impl Future<Output = Result<Option<RpcTransaction<Self::NetworkTypes>>, Self::Error>> + Send
396 where
397 Self: LoadBlock + LoadState,
398 {
399 async move {
400 if include_pending &&
402 let Some(tx) =
403 RpcNodeCore::pool(self).get_transaction_by_sender_and_nonce(sender, nonce)
404 {
405 let transaction = tx.transaction.clone_into_consensus();
406 return Ok(Some(self.converter().fill_pending(transaction)?));
407 }
408
409 let highest = self.transaction_count(sender, None).await?.saturating_to::<u64>();
413
414 if nonce >= highest {
417 return Ok(None);
418 }
419
420 let high = self.provider().best_block_number().map_err(Self::Error::from_eth_err)?;
421
422 let num = binary_search::<_, _, Self::Error>(1, high, |mid| async move {
425 let mid_nonce =
426 self.transaction_count(sender, Some(mid.into())).await?.saturating_to::<u64>();
427
428 Ok(mid_nonce > nonce)
429 })
430 .await?;
431
432 let block_id = num.into();
433 self.recovered_block(block_id)
434 .await?
435 .and_then(|block| {
436 let block_hash = block.hash();
437 let block_number = block.number();
438 let block_timestamp = block.timestamp();
439 let base_fee_per_gas = block.base_fee_per_gas();
440
441 block
442 .transactions_with_sender()
443 .enumerate()
444 .find(|(_, (signer, tx))| **signer == sender && (*tx).nonce() == nonce)
445 .map(|(index, (signer, tx))| {
446 let tx_info = TransactionInfo {
447 hash: Some(*tx.tx_hash()),
448 block_hash: Some(block_hash),
449 block_number: Some(block_number),
450 block_timestamp: Some(block_timestamp),
451 base_fee: base_fee_per_gas,
452 index: Some(index as u64),
453 };
454 Ok(self.converter().fill(tx.clone().with_signer(*signer), tx_info)?)
455 })
456 })
457 .ok_or(EthApiError::HeaderNotFound(block_id))?
458 .map(Some)
459 }
460 }
461
462 fn raw_transaction_by_block_and_tx_index(
466 &self,
467 block_id: BlockId,
468 index: usize,
469 ) -> impl Future<Output = Result<Option<Bytes>, Self::Error>> + Send
470 where
471 Self: LoadBlock,
472 {
473 async move {
474 if let Some(block) = self.recovered_block(block_id).await? &&
475 let Some(tx) = block.body().transactions().get(index)
476 {
477 return Ok(Some(tx.encoded_2718().into()))
478 }
479
480 Ok(None)
481 }
482 }
483
484 fn send_transaction_request(
487 &self,
488 mut request: RpcTxReq<Self::NetworkTypes>,
489 ) -> impl Future<Output = Result<B256, Self::Error>> + Send
490 where
491 Self: EthApiSpec + LoadBlock + EstimateCall,
492 {
493 async move {
494 let from = match request.as_ref().from() {
495 Some(from) => from,
496 None => return Err(SignError::NoAccount.into_eth_err()),
497 };
498
499 if self.find_signer(&from).is_err() {
500 return Err(SignError::NoAccount.into_eth_err())
501 }
502
503 if request.as_ref().nonce().is_none() {
505 let nonce = self.next_available_nonce_for(&request).await?;
506 request.as_mut().set_nonce(nonce);
507 }
508
509 let chain_id = self.chain_id();
510 request.as_mut().set_chain_id(chain_id.to());
511
512 if request.as_ref().gas_limit().is_none() {
513 let estimated_gas = self
514 .estimate_gas_at(request.clone(), BlockId::pending(), EvmOverrides::default())
515 .await?;
516 request.as_mut().set_gas_limit(estimated_gas.to());
517 }
518
519 let transaction = self.sign_request(&from, request).await?.with_signer(from);
520
521 let pool_transaction =
522 <<Self as RpcNodeCore>::Pool as TransactionPool>::Transaction::try_from_consensus(
523 transaction,
524 )
525 .map_err(|e| {
526 Self::Error::from_eth_err(TransactionConversionError::Other(e.to_string()))
527 })?;
528
529 let AddedTransactionOutcome { hash, .. } = self
531 .pool()
532 .add_transaction(TransactionOrigin::Local, pool_transaction)
533 .await
534 .map_err(Self::Error::from_eth_err)?;
535
536 Ok(hash)
537 }
538 }
539
540 fn fill_transaction(
542 &self,
543 mut request: RpcTxReq<Self::NetworkTypes>,
544 ) -> impl Future<Output = Result<FillTransaction<TxTy<Self::Primitives>>, Self::Error>> + Send
545 where
546 Self: EthApiSpec + LoadBlock + EstimateCall + LoadFee,
547 {
548 async move {
549 if request.as_ref().value().is_none() {
550 request.as_mut().set_value(U256::ZERO);
551 }
552
553 if request.as_ref().nonce().is_none() {
554 let nonce = self.next_available_nonce_for(&request).await?;
555 request.as_mut().set_nonce(nonce);
556 }
557
558 let chain_id = self.chain_id();
559 request.as_mut().set_chain_id(chain_id.to());
560
561 if request.as_ref().has_eip4844_fields() &&
562 request.as_ref().max_fee_per_blob_gas().is_none()
563 {
564 let blob_fee = self.blob_base_fee().await?;
565 request.as_mut().set_max_fee_per_blob_gas(blob_fee.to());
566 }
567
568 if request.as_ref().sidecar.is_some() &&
571 request.as_ref().blob_versioned_hashes.is_none()
572 {
573 request.as_mut().populate_blob_hashes();
574 }
575
576 if request.as_ref().gas_limit().is_none() {
577 let estimated_gas = self
578 .estimate_gas_at(request.clone(), BlockId::pending(), EvmOverrides::default())
579 .await?;
580 request.as_mut().set_gas_limit(estimated_gas.to());
581 }
582
583 if request.as_ref().gas_price().is_none() {
584 let tip = if let Some(tip) = request.as_ref().max_priority_fee_per_gas() {
585 tip
586 } else {
587 let tip = self.suggested_priority_fee().await?.to::<u128>();
588 request.as_mut().set_max_priority_fee_per_gas(tip);
589 tip
590 };
591 if request.as_ref().max_fee_per_gas().is_none() {
592 let header =
593 self.provider().latest_header().map_err(Self::Error::from_eth_err)?;
594 let base_fee = header.and_then(|h| h.base_fee_per_gas()).unwrap_or_default();
595 request.as_mut().set_max_fee_per_gas(base_fee as u128 * 2 + tip);
602 }
603 }
604
605 let tx = self.converter().build_simulate_v1_transaction(request)?;
606
607 let raw = tx.encoded_2718().into();
608
609 Ok(FillTransaction { raw, tx })
610 }
611 }
612
613 fn sign_request(
615 &self,
616 from: &Address,
617 txn: RpcTxReq<Self::NetworkTypes>,
618 ) -> impl Future<Output = Result<ProviderTx<Self::Provider>, Self::Error>> + Send {
619 async move {
620 self.find_signer(from)?
621 .sign_transaction(txn, from)
622 .await
623 .map_err(Self::Error::from_eth_err)
624 }
625 }
626
627 fn sign(
629 &self,
630 account: Address,
631 message: Bytes,
632 ) -> impl Future<Output = Result<Bytes, Self::Error>> + Send {
633 async move {
634 Ok(self
635 .find_signer(&account)?
636 .sign(account, &message)
637 .await
638 .map_err(Self::Error::from_eth_err)?
639 .as_bytes()
640 .into())
641 }
642 }
643
644 fn sign_transaction(
647 &self,
648 request: RpcTxReq<Self::NetworkTypes>,
649 ) -> impl Future<Output = Result<Bytes, Self::Error>> + Send {
650 async move {
651 let from = match request.as_ref().from() {
652 Some(from) => from,
653 None => return Err(SignError::NoAccount.into_eth_err()),
654 };
655
656 Ok(self.sign_request(&from, request).await?.encoded_2718().into())
657 }
658 }
659
660 fn sign_typed_data(&self, data: &TypedData, account: Address) -> Result<Bytes, Self::Error> {
662 Ok(self
663 .find_signer(&account)?
664 .sign_typed_data(account, data)
665 .map_err(Self::Error::from_eth_err)?
666 .as_bytes()
667 .into())
668 }
669
670 #[expect(clippy::type_complexity)]
672 fn find_signer(
673 &self,
674 account: &Address,
675 ) -> Result<
676 Box<dyn EthSigner<ProviderTx<Self::Provider>, RpcTxReq<Self::NetworkTypes>> + 'static>,
677 Self::Error,
678 > {
679 self.signers()
680 .read()
681 .iter()
682 .find(|signer| signer.is_signer_for(account))
683 .map(|signer| dyn_clone::clone_box(&**signer))
684 .ok_or_else(|| SignError::NoAccount.into_eth_err())
685 }
686}
687
688pub trait LoadTransaction: SpawnBlocking + FullEthApiTypes + RpcNodeCoreExt {
693 #[expect(clippy::complexity)]
699 fn transaction_by_hash(
700 &self,
701 hash: B256,
702 ) -> impl Future<
703 Output = Result<Option<TransactionSource<ProviderTx<Self::Provider>>>, Self::Error>,
704 > + Send {
705 async move {
706 if let Some(cached) = self.cache().get_transaction_by_hash(hash).await &&
708 let Some(source) = cached.to_transaction_source()
709 {
710 return Ok(Some(source));
711 }
712
713 if let Some((tx, meta)) = self
715 .spawn_blocking_io(move |this| {
716 this.provider()
717 .transaction_by_hash_with_meta(hash)
718 .map_err(Self::Error::from_eth_err)
719 })
720 .await?
721 {
722 let transaction = tx
726 .try_into_recovered_unchecked()
727 .map_err(|_| EthApiError::InvalidTransactionSignature)?;
728
729 return Ok(Some(TransactionSource::Block {
730 transaction,
731 index: meta.index,
732 block_hash: meta.block_hash,
733 block_number: meta.block_number,
734 block_timestamp: meta.timestamp,
735 base_fee: meta.base_fee,
736 }));
737 }
738
739 if let Some(tx) = self.pool().get(&hash).map(|tx| tx.transaction.clone_into_consensus())
741 {
742 return Ok(Some(TransactionSource::Pool(tx.into())));
743 }
744
745 Ok(None)
746 }
747 }
748
749 #[expect(clippy::type_complexity)]
753 fn transaction_by_hash_at(
754 &self,
755 transaction_hash: B256,
756 ) -> impl Future<
757 Output = Result<
758 Option<(TransactionSource<ProviderTx<Self::Provider>>, BlockId)>,
759 Self::Error,
760 >,
761 > + Send {
762 async move {
763 Ok(self.transaction_by_hash(transaction_hash).await?.map(|tx| match tx {
764 tx @ TransactionSource::Pool(_) => (tx, BlockId::pending()),
765 tx @ TransactionSource::Block { block_hash, .. } => {
766 (tx, BlockId::Hash(block_hash.into()))
767 }
768 }))
769 }
770 }
771
772 #[expect(clippy::type_complexity)]
774 fn transaction_and_block(
775 &self,
776 hash: B256,
777 ) -> impl Future<
778 Output = Result<
779 Option<(
780 TransactionSource<ProviderTx<Self::Provider>>,
781 Arc<RecoveredBlock<ProviderBlock<Self::Provider>>>,
782 )>,
783 Self::Error,
784 >,
785 > + Send {
786 async move {
787 let (transaction, at) = match self.transaction_by_hash_at(hash).await? {
788 None => return Ok(None),
789 Some(res) => res,
790 };
791
792 let block_hash = match at {
794 BlockId::Hash(hash) => hash.block_hash,
795 _ => return Ok(None),
796 };
797 let block = self
798 .cache()
799 .get_recovered_block(block_hash)
800 .await
801 .map_err(Self::Error::from_eth_err)?;
802 Ok(block.map(|block| (transaction, block)))
803 }
804 }
805}