-
Notifications
You must be signed in to change notification settings - Fork 892
/
traits.rs
972 lines (843 loc) · 35.1 KB
/
traits.rs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
use crate::{
error::PoolResult,
pool::{state::SubPool, TransactionEvents},
validate::ValidPoolTransaction,
AllTransactionsEvents,
};
use futures_util::{ready, Stream};
use reth_primitives::{
Address, BlobTransactionSidecar, BlobTransactionValidationError,
FromRecoveredPooledTransaction, FromRecoveredTransaction, IntoRecoveredTransaction, PeerId,
PooledTransactionsElement, PooledTransactionsElementEcRecovered, SealedBlock, Transaction,
TransactionKind, TransactionSignedEcRecovered, TxEip4844, TxHash, EIP1559_TX_TYPE_ID,
EIP4844_TX_TYPE_ID, H256, U256,
};
use reth_rlp::Encodable;
use std::{
collections::{HashMap, HashSet},
fmt,
pin::Pin,
sync::Arc,
task::{Context, Poll},
};
use tokio::sync::mpsc::Receiver;
use crate::blobstore::BlobStoreError;
use reth_primitives::kzg::KzgSettings;
#[cfg(feature = "serde")]
use serde::{Deserialize, Serialize};
/// General purpose abstraction of a transaction-pool.
///
/// This is intended to be used by API-consumers such as RPC that need inject new incoming,
/// unverified transactions. And by block production that needs to get transactions to execute in a
/// new block.
///
/// Note: This requires `Clone` for convenience, since it is assumed that this will be implemented
/// for a wrapped `Arc` type, see also [`Pool`](crate::Pool).
#[async_trait::async_trait]
#[auto_impl::auto_impl(Arc)]
pub trait TransactionPool: Send + Sync + Clone {
/// The transaction type of the pool
type Transaction: PoolTransaction;
/// Returns stats about the pool and all sub-pools.
fn pool_size(&self) -> PoolSize;
/// Returns the block the pool is currently tracking.
///
/// This tracks the block that the pool has last seen.
fn block_info(&self) -> BlockInfo;
/// Imports an _external_ transaction.
///
/// This is intended to be used by the network to insert incoming transactions received over the
/// p2p network.
///
/// Consumer: P2P
async fn add_external_transaction(&self, transaction: Self::Transaction) -> PoolResult<TxHash> {
self.add_transaction(TransactionOrigin::External, transaction).await
}
/// Imports all _external_ transactions
///
///
/// Consumer: Utility
async fn add_external_transactions(
&self,
transactions: Vec<Self::Transaction>,
) -> PoolResult<Vec<PoolResult<TxHash>>> {
self.add_transactions(TransactionOrigin::External, transactions).await
}
/// Adds an _unvalidated_ transaction into the pool and subscribe to state changes.
///
/// This is the same as [TransactionPool::add_transaction] but returns an event stream for the
/// given transaction.
///
/// Consumer: Custom
async fn add_transaction_and_subscribe(
&self,
origin: TransactionOrigin,
transaction: Self::Transaction,
) -> PoolResult<TransactionEvents>;
/// Adds an _unvalidated_ transaction into the pool.
///
/// Consumer: RPC
async fn add_transaction(
&self,
origin: TransactionOrigin,
transaction: Self::Transaction,
) -> PoolResult<TxHash>;
/// Adds the given _unvalidated_ transaction into the pool.
///
/// Returns a list of results.
///
/// Consumer: RPC
async fn add_transactions(
&self,
origin: TransactionOrigin,
transactions: Vec<Self::Transaction>,
) -> PoolResult<Vec<PoolResult<TxHash>>>;
/// Returns a new transaction change event stream for the given transaction.
///
/// Returns `None` if the transaction is not in the pool.
fn transaction_event_listener(&self, tx_hash: TxHash) -> Option<TransactionEvents>;
/// Returns a new transaction change event stream for _all_ transactions in the pool.
fn all_transactions_event_listener(&self) -> AllTransactionsEvents<Self::Transaction>;
/// Returns a new Stream that yields transactions hashes for new __pending__ transactions
/// inserted into the pool that are allowed to be propagated.
///
/// Note: This is intended for networking and will __only__ yield transactions that are allowed
/// to be propagated over the network.
///
/// Consumer: RPC/P2P
fn pending_transactions_listener(&self) -> Receiver<TxHash> {
self.pending_transactions_listener_for(TransactionListenerKind::PropagateOnly)
}
/// Returns a new Stream that yields transactions hashes for new __pending__ transactions
/// inserted into the pool depending on the given [TransactionListenerKind] argument.
fn pending_transactions_listener_for(&self, kind: TransactionListenerKind) -> Receiver<TxHash>;
/// Returns a new stream that yields new valid transactions added to the pool.
fn new_transactions_listener(&self) -> Receiver<NewTransactionEvent<Self::Transaction>> {
self.new_transactions_listener_for(TransactionListenerKind::PropagateOnly)
}
/// Returns a new stream that yields new valid transactions added to the pool
/// depending on the given [TransactionListenerKind] argument.
fn new_transactions_listener_for(
&self,
kind: TransactionListenerKind,
) -> Receiver<NewTransactionEvent<Self::Transaction>>;
/// Returns a new Stream that yields new transactions added to the pending sub-pool.
///
/// This is a convenience wrapper around [Self::new_transactions_listener] that filters for
/// [SubPool::Pending](crate::SubPool).
fn new_pending_pool_transactions_listener(
&self,
) -> NewSubpoolTransactionStream<Self::Transaction> {
NewSubpoolTransactionStream::new(
self.new_transactions_listener_for(TransactionListenerKind::PropagateOnly),
SubPool::Pending,
)
}
/// Returns a new Stream that yields new transactions added to the basefee sub-pool.
///
/// This is a convenience wrapper around [Self::new_transactions_listener] that filters for
/// [SubPool::BaseFee](crate::SubPool).
fn new_basefee_pool_transactions_listener(
&self,
) -> NewSubpoolTransactionStream<Self::Transaction> {
NewSubpoolTransactionStream::new(self.new_transactions_listener(), SubPool::BaseFee)
}
/// Returns a new Stream that yields new transactions added to the queued-pool.
///
/// This is a convenience wrapper around [Self::new_transactions_listener] that filters for
/// [SubPool::Queued](crate::SubPool).
fn new_queued_transactions_listener(&self) -> NewSubpoolTransactionStream<Self::Transaction> {
NewSubpoolTransactionStream::new(self.new_transactions_listener(), SubPool::Queued)
}
/// Returns the _hashes_ of all transactions in the pool.
///
/// Note: This returns a `Vec` but should guarantee that all hashes are unique.
///
/// Consumer: P2P
fn pooled_transaction_hashes(&self) -> Vec<TxHash>;
/// Returns only the first `max` hashes of transactions in the pool.
///
/// Consumer: P2P
fn pooled_transaction_hashes_max(&self, max: usize) -> Vec<TxHash>;
/// Returns the _full_ transaction objects all transactions in the pool.
///
/// This is intended to be used by the network for the initial exchange of pooled transaction
/// _hashes_
///
/// Note: This returns a `Vec` but should guarantee that all transactions are unique.
///
/// Caution: In case of blob transactions, this does not include the sidecar.
///
/// Consumer: P2P
fn pooled_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Returns only the first `max` transactions in the pool.
///
/// Consumer: P2P
fn pooled_transactions_max(
&self,
max: usize,
) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Returns converted [PooledTransactionsElement] for the given transaction hashes.
///
/// This adheres to the expected behavior of [`GetPooledTransactions`](https://github.com/ethereum/devp2p/blob/master/caps/eth.md#getpooledtransactions-0x09):
/// The transactions must be in same order as in the request, but it is OK to skip transactions
/// which are not available.
///
/// If the transaction is a blob transaction, the sidecar will be included.
///
/// Consumer: P2P
fn get_pooled_transaction_elements(
&self,
tx_hashes: Vec<TxHash>,
limit: GetPooledTransactionLimit,
) -> Vec<PooledTransactionsElement>;
/// Returns an iterator that yields transactions that are ready for block production.
///
/// Consumer: Block production
fn best_transactions(
&self,
) -> Box<dyn BestTransactions<Item = Arc<ValidPoolTransaction<Self::Transaction>>>>;
/// Returns an iterator that yields transactions that are ready for block production with the
/// given base fee.
///
/// Consumer: Block production
fn best_transactions_with_base_fee(
&self,
base_fee: u64,
) -> Box<dyn BestTransactions<Item = Arc<ValidPoolTransaction<Self::Transaction>>>>;
/// Returns all transactions that can be included in the next block.
///
/// This is primarily used for the `txpool_` RPC namespace: <https://geth.ethereum.org/docs/interacting-with-geth/rpc/ns-txpool> which distinguishes between `pending` and `queued` transactions, where `pending` are transactions ready for inclusion in the next block and `queued` are transactions that are ready for inclusion in future blocks.
///
/// Consumer: RPC
fn pending_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Returns all transactions that can be included in _future_ blocks.
///
/// This and [Self::pending_transactions] are mutually exclusive.
///
/// Consumer: RPC
fn queued_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Returns all transactions that are currently in the pool grouped by whether they are ready
/// for inclusion in the next block or not.
///
/// This is primarily used for the `txpool_` namespace: <https://geth.ethereum.org/docs/interacting-with-geth/rpc/ns-txpool>
///
/// Consumer: RPC
fn all_transactions(&self) -> AllPoolTransactions<Self::Transaction>;
/// Removes all transactions corresponding to the given hashes.
///
/// Also removes all _dependent_ transactions.
///
/// Consumer: Block production
fn remove_transactions(
&self,
hashes: Vec<TxHash>,
) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Retains only those hashes that are unknown to the pool.
/// In other words, removes all transactions from the given set that are currently present in
/// the pool.
///
/// Consumer: P2P
fn retain_unknown(&self, hashes: &mut Vec<TxHash>);
/// Returns if the transaction for the given hash is already included in this pool.
fn contains(&self, tx_hash: &TxHash) -> bool {
self.get(tx_hash).is_some()
}
/// Returns the transaction for the given hash.
fn get(&self, tx_hash: &TxHash) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Returns all transactions objects for the given hashes.
///
/// Caution: This in case of blob transactions, this does not include the sidecar.
fn get_all(&self, txs: Vec<TxHash>) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Notify the pool about transactions that are propagated to peers.
///
/// Consumer: P2P
fn on_propagated(&self, txs: PropagatedTransactions);
/// Returns all transactions sent by a given user
fn get_transactions_by_sender(
&self,
sender: Address,
) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>>;
/// Returns a set of all senders of transactions in the pool
fn unique_senders(&self) -> HashSet<Address>;
/// Returns the [BlobTransactionSidecar] for the given transaction hash if it exists in the blob
/// store.
fn get_blob(&self, tx_hash: TxHash) -> Result<Option<BlobTransactionSidecar>, BlobStoreError>;
/// Returns all [BlobTransactionSidecar] for the given transaction hashes if they exists in the
/// blob store.
///
/// This only returns the blobs that were found in the store.
/// If there's no blob it will not be returned.
fn get_all_blobs(
&self,
tx_hashes: Vec<TxHash>,
) -> Result<Vec<(TxHash, BlobTransactionSidecar)>, BlobStoreError>;
}
/// Extension for [TransactionPool] trait that allows to set the current block info.
#[auto_impl::auto_impl(Arc)]
pub trait TransactionPoolExt: TransactionPool {
/// Sets the current block info for the pool.
fn set_block_info(&self, info: BlockInfo);
/// Event listener for when the pool needs to be updated
///
/// Implementers need to update the pool accordingly.
/// For example the base fee of the pending block is determined after a block is mined which
/// affects the dynamic fee requirement of pending transactions in the pool.
fn on_canonical_state_change(&self, update: CanonicalStateUpdate<'_>);
/// Updates the accounts in the pool
fn update_accounts(&self, accounts: Vec<ChangedAccount>);
/// Deletes the blob sidecar for the given transaction from the blob store
fn delete_blob(&self, tx: H256);
/// Deletes multiple blob sidecars from the blob store
fn delete_blobs(&self, txs: Vec<H256>);
}
/// Determines what kind of new transactions should be emitted by a stream of transactions.
///
/// This gives control whether to include transactions that are allowed to be propagated.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum TransactionListenerKind {
/// Any new pending transactions
All,
/// Only transactions that are allowed to be propagated.
///
/// See also [ValidPoolTransaction]
PropagateOnly,
}
impl TransactionListenerKind {
/// Returns true if we're only interested in transactions that are allowed to be propagated.
#[inline]
pub fn is_propagate_only(&self) -> bool {
matches!(self, Self::PropagateOnly)
}
}
/// A Helper type that bundles all transactions in the pool.
#[derive(Debug, Clone)]
pub struct AllPoolTransactions<T: PoolTransaction> {
/// Transactions that are ready for inclusion in the next block.
pub pending: Vec<Arc<ValidPoolTransaction<T>>>,
/// Transactions that are ready for inclusion in _future_ blocks, but are currently parked,
/// because they depend on other transactions that are not yet included in the pool (nonce gap)
/// or otherwise blocked.
pub queued: Vec<Arc<ValidPoolTransaction<T>>>,
}
// === impl AllPoolTransactions ===
impl<T: PoolTransaction> AllPoolTransactions<T> {
/// Returns an iterator over all pending [TransactionSignedEcRecovered] transactions.
pub fn pending_recovered(&self) -> impl Iterator<Item = TransactionSignedEcRecovered> + '_ {
self.pending.iter().map(|tx| tx.transaction.to_recovered_transaction())
}
/// Returns an iterator over all queued [TransactionSignedEcRecovered] transactions.
pub fn queued_recovered(&self) -> impl Iterator<Item = TransactionSignedEcRecovered> + '_ {
self.queued.iter().map(|tx| tx.transaction.to_recovered_transaction())
}
}
impl<T: PoolTransaction> Default for AllPoolTransactions<T> {
fn default() -> Self {
Self { pending: Default::default(), queued: Default::default() }
}
}
/// Represents a transaction that was propagated over the network.
#[derive(Debug, Clone, Eq, PartialEq, Default)]
pub struct PropagatedTransactions(pub HashMap<TxHash, Vec<PropagateKind>>);
/// Represents how a transaction was propagated over the network.
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
pub enum PropagateKind {
/// The full transaction object was sent to the peer.
///
/// This is equivalent to the `Transaction` message
Full(PeerId),
/// Only the Hash was propagated to the peer.
Hash(PeerId),
}
// === impl PropagateKind ===
impl PropagateKind {
/// Returns the peer the transaction was sent to
pub fn peer(&self) -> &PeerId {
match self {
PropagateKind::Full(peer) => peer,
PropagateKind::Hash(peer) => peer,
}
}
}
impl From<PropagateKind> for PeerId {
fn from(value: PropagateKind) -> Self {
match value {
PropagateKind::Full(peer) => peer,
PropagateKind::Hash(peer) => peer,
}
}
}
/// Represents a new transaction
#[derive(Debug)]
pub struct NewTransactionEvent<T: PoolTransaction> {
/// The pool which the transaction was moved to.
pub subpool: SubPool,
/// Actual transaction
pub transaction: Arc<ValidPoolTransaction<T>>,
}
impl<T: PoolTransaction> Clone for NewTransactionEvent<T> {
fn clone(&self) -> Self {
Self { subpool: self.subpool, transaction: self.transaction.clone() }
}
}
/// Where the transaction originates from.
///
/// Depending on where the transaction was picked up, it affects how the transaction is handled
/// internally, e.g. limits for simultaneous transaction of one sender.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum TransactionOrigin {
/// Transaction is coming from a local source.
Local,
/// Transaction has been received externally.
///
/// This is usually considered an "untrusted" source, for example received from another in the
/// network.
External,
/// Transaction is originated locally and is intended to remain private.
///
/// This type of transaction should not be propagated to the network. It's meant for
/// private usage within the local node only.
Private,
}
// === impl TransactionOrigin ===
impl TransactionOrigin {
/// Whether the transaction originates from a local source.
pub fn is_local(&self) -> bool {
matches!(self, TransactionOrigin::Local)
}
}
/// Represents changes after a new canonical block or range of canonical blocks was added to the
/// chain.
///
/// It is expected that this is only used if the added blocks are canonical to the pool's last known
/// block hash. In other words, the first added block of the range must be the child of the last
/// known block hash.
///
/// This is used to update the pool state accordingly.
#[derive(Debug, Clone)]
pub struct CanonicalStateUpdate<'a> {
/// Hash of the tip block.
pub new_tip: &'a SealedBlock,
/// EIP-1559 Base fee of the _next_ (pending) block
///
/// The base fee of a block depends on the utilization of the last block and its base fee.
pub pending_block_base_fee: u64,
/// A set of changed accounts across a range of blocks.
pub changed_accounts: Vec<ChangedAccount>,
/// All mined transactions in the block range.
pub mined_transactions: Vec<H256>,
}
impl<'a> CanonicalStateUpdate<'a> {
/// Returns the number of the tip block.
pub fn number(&self) -> u64 {
self.new_tip.number
}
/// Returns the hash of the tip block.
pub fn hash(&self) -> H256 {
self.new_tip.hash
}
/// Timestamp of the latest chain update
pub fn timestamp(&self) -> u64 {
self.new_tip.timestamp
}
/// Returns the block info for the tip block.
pub fn block_info(&self) -> BlockInfo {
BlockInfo {
last_seen_block_hash: self.hash(),
last_seen_block_number: self.number(),
pending_basefee: self.pending_block_base_fee,
}
}
}
impl<'a> fmt::Display for CanonicalStateUpdate<'a> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{{ hash: {}, number: {}, pending_block_base_fee: {}, changed_accounts: {}, mined_transactions: {} }}",
self.hash(), self.number(), self.pending_block_base_fee, self.changed_accounts.len(), self.mined_transactions.len())
}
}
/// Represents a changed account
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub struct ChangedAccount {
/// The address of the account.
pub address: Address,
/// Account nonce.
pub nonce: u64,
/// Account balance.
pub balance: U256,
}
// === impl ChangedAccount ===
impl ChangedAccount {
/// Creates a new `ChangedAccount` with the given address and 0 balance and nonce.
pub(crate) fn empty(address: Address) -> Self {
Self { address, nonce: 0, balance: U256::ZERO }
}
}
/// An `Iterator` that only returns transactions that are ready to be executed.
///
/// This makes no assumptions about the order of the transactions, but expects that _all_
/// transactions are valid (no nonce gaps.) for the tracked state of the pool.
///
/// Note: this iterator will always return the best transaction that it currently knows.
/// There is no guarantee transactions will be returned sequentially in decreasing
/// priority order.
pub trait BestTransactions: Iterator + Send {
/// Mark the transaction as invalid.
///
/// Implementers must ensure all subsequent transaction _don't_ depend on this transaction.
/// In other words, this must remove the given transaction _and_ drain all transaction that
/// depend on it.
fn mark_invalid(&mut self, transaction: &Self::Item);
/// An iterator may be able to receive additional pending transactions that weren't present it
/// the pool when it was created.
///
/// This ensures that iterator will return the best transaction that it currently knows and not
/// listen to pool updates.
fn no_updates(&mut self);
}
/// A no-op implementation that yields no transactions.
impl<T> BestTransactions for std::iter::Empty<T> {
fn mark_invalid(&mut self, _tx: &T) {}
fn no_updates(&mut self) {}
}
/// Trait for transaction types used inside the pool
pub trait PoolTransaction:
fmt::Debug
+ Send
+ Sync
+ FromRecoveredPooledTransaction
+ FromRecoveredTransaction
+ IntoRecoveredTransaction
{
/// Hash of the transaction.
fn hash(&self) -> &TxHash;
/// The Sender of the transaction.
fn sender(&self) -> Address;
/// Returns the nonce for this transaction.
fn nonce(&self) -> u64;
/// Returns the cost that this transaction is allowed to consume:
///
/// For EIP-1559 transactions: `max_fee_per_gas * gas_limit + tx_value`.
/// For legacy transactions: `gas_price * gas_limit + tx_value`.
fn cost(&self) -> U256;
/// Amount of gas that should be used in executing this transaction. This is paid up-front.
fn gas_limit(&self) -> u64;
/// Returns the EIP-1559 the maximum fee per gas the caller is willing to pay.
///
/// For legacy transactions this is gas_price.
///
/// This is also commonly referred to as the "Gas Fee Cap" (`GasFeeCap`).
fn max_fee_per_gas(&self) -> u128;
/// Returns the EIP-1559 Priority fee the caller is paying to the block author.
///
/// This will return `None` for non-EIP1559 transactions
fn max_priority_fee_per_gas(&self) -> Option<u128>;
/// Returns the effective tip for this transaction.
///
/// For EIP-1559 transactions: `min(max_fee_per_gas - base_fee, max_priority_fee_per_gas)`.
/// For legacy transactions: `gas_price - base_fee`.
fn effective_tip_per_gas(&self, base_fee: u64) -> Option<u128>;
/// Returns the max priority fee per gas if the transaction is an EIP-1559 transaction, and
/// otherwise returns the gas price.
fn priority_fee_or_price(&self) -> u128;
/// Returns the transaction's [`TransactionKind`], which is the address of the recipient or
/// [`TransactionKind::Create`] if the transaction is a contract creation.
fn kind(&self) -> &TransactionKind;
/// Returns a measurement of the heap usage of this type and all its internals.
fn size(&self) -> usize;
/// Returns the transaction type
fn tx_type(&self) -> u8;
/// Returns true if the transaction is an EIP-1559 transaction.
fn is_eip1559(&self) -> bool {
self.tx_type() == EIP1559_TX_TYPE_ID
}
/// Returns true if the transaction is an EIP-4844 transaction.
fn is_eip4844(&self) -> bool {
self.tx_type() == EIP4844_TX_TYPE_ID
}
/// Returns the length of the rlp encoded object
fn encoded_length(&self) -> usize;
/// Returns chain_id
fn chain_id(&self) -> Option<u64>;
}
/// An extension trait that provides additional interfaces for the
/// [EthTransactionValidator](crate::EthTransactionValidator).
pub trait EthPoolTransaction: PoolTransaction {
/// Extracts the blob sidecar from the transaction.
fn take_blob(&mut self) -> EthBlobTransactionSidecar;
/// Returns the transaction as EIP-4844 transaction if it is one.
fn as_eip4844(&self) -> Option<&TxEip4844>;
/// Validates the blob sidecar of the transaction with the given settings.
fn validate_blob(
&self,
blob: &BlobTransactionSidecar,
settings: &KzgSettings,
) -> Result<(), BlobTransactionValidationError>;
}
/// The default [PoolTransaction] for the [Pool](crate::Pool) for Ethereum.
///
/// This type is essentially a wrapper around [TransactionSignedEcRecovered] with additional fields
/// derived from the transaction that are frequently used by the pools for ordering.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EthPooledTransaction {
/// EcRecovered transaction info
pub(crate) transaction: TransactionSignedEcRecovered,
/// For EIP-1559 transactions: `max_fee_per_gas * gas_limit + tx_value`.
/// For legacy transactions: `gas_price * gas_limit + tx_value`.
pub(crate) cost: U256,
/// The blob side car this transaction
pub(crate) blob_sidecar: EthBlobTransactionSidecar,
}
/// Represents the blob sidecar of the [EthPooledTransaction].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EthBlobTransactionSidecar {
/// This transaction does not have a blob sidecar
None,
/// This transaction has a blob sidecar (EIP-4844) but it is missing
///
/// It was either extracted after being inserted into the pool or re-injected after reorg
/// without the blob sidecar
Missing,
/// The eip-4844 transaction was pulled from the network and still has its blob sidecar
Present(BlobTransactionSidecar),
}
impl EthPooledTransaction {
/// Create new instance of [Self].
pub fn new(transaction: TransactionSignedEcRecovered) -> Self {
let mut blob_sidecar = EthBlobTransactionSidecar::None;
let gas_cost = match &transaction.transaction {
Transaction::Legacy(t) => U256::from(t.gas_price) * U256::from(t.gas_limit),
Transaction::Eip2930(t) => U256::from(t.gas_price) * U256::from(t.gas_limit),
Transaction::Eip1559(t) => U256::from(t.max_fee_per_gas) * U256::from(t.gas_limit),
Transaction::Eip4844(t) => {
blob_sidecar = EthBlobTransactionSidecar::Missing;
U256::from(t.max_fee_per_gas) * U256::from(t.gas_limit)
}
};
let cost = gas_cost + U256::from(transaction.value());
Self { transaction, cost, blob_sidecar }
}
/// Return the reference to the underlying transaction.
pub fn transaction(&self) -> &TransactionSignedEcRecovered {
&self.transaction
}
}
/// Conversion from the network transaction type to the pool transaction type.
impl From<PooledTransactionsElementEcRecovered> for EthPooledTransaction {
fn from(tx: PooledTransactionsElementEcRecovered) -> Self {
let (tx, signer) = tx.into_components();
match tx {
PooledTransactionsElement::BlobTransaction(tx) => {
// include the blob sidecar
let (tx, blob) = tx.into_parts();
let tx = TransactionSignedEcRecovered::from_signed_transaction(tx, signer);
let mut pooled = EthPooledTransaction::new(tx);
pooled.blob_sidecar = EthBlobTransactionSidecar::Present(blob);
pooled
}
tx => {
// no blob sidecar
EthPooledTransaction::new(tx.into_ecrecovered_transaction(signer))
}
}
}
}
impl PoolTransaction for EthPooledTransaction {
/// Returns hash of the transaction.
fn hash(&self) -> &TxHash {
self.transaction.hash_ref()
}
/// Returns the Sender of the transaction.
fn sender(&self) -> Address {
self.transaction.signer()
}
/// Returns the nonce for this transaction.
fn nonce(&self) -> u64 {
self.transaction.nonce()
}
/// Returns the cost that this transaction is allowed to consume:
///
/// For EIP-1559 transactions: `max_fee_per_gas * gas_limit + tx_value`.
/// For legacy transactions: `gas_price * gas_limit + tx_value`.
fn cost(&self) -> U256 {
self.cost
}
/// Amount of gas that should be used in executing this transaction. This is paid up-front.
fn gas_limit(&self) -> u64 {
self.transaction.gas_limit()
}
/// Returns the EIP-1559 Max base fee the caller is willing to pay.
///
/// For legacy transactions this is gas_price.
///
/// This is also commonly referred to as the "Gas Fee Cap" (`GasFeeCap`).
fn max_fee_per_gas(&self) -> u128 {
match &self.transaction.transaction {
Transaction::Legacy(tx) => tx.gas_price,
Transaction::Eip2930(tx) => tx.gas_price,
Transaction::Eip1559(tx) => tx.max_fee_per_gas,
Transaction::Eip4844(tx) => tx.max_fee_per_gas,
}
}
/// Returns the EIP-1559 Priority fee the caller is paying to the block author.
///
/// This will return `None` for non-EIP1559 transactions
fn max_priority_fee_per_gas(&self) -> Option<u128> {
match &self.transaction.transaction {
Transaction::Legacy(_) => None,
Transaction::Eip2930(_) => None,
Transaction::Eip1559(tx) => Some(tx.max_priority_fee_per_gas),
Transaction::Eip4844(tx) => Some(tx.max_priority_fee_per_gas),
}
}
/// Returns the effective tip for this transaction.
///
/// For EIP-1559 transactions: `min(max_fee_per_gas - base_fee, max_priority_fee_per_gas)`.
/// For legacy transactions: `gas_price - base_fee`.
fn effective_tip_per_gas(&self, base_fee: u64) -> Option<u128> {
self.transaction.effective_tip_per_gas(base_fee)
}
/// Returns the max priority fee per gas if the transaction is an EIP-1559 transaction, and
/// otherwise returns the gas price.
fn priority_fee_or_price(&self) -> u128 {
self.transaction.priority_fee_or_price()
}
/// Returns the transaction's [`TransactionKind`], which is the address of the recipient or
/// [`TransactionKind::Create`] if the transaction is a contract creation.
fn kind(&self) -> &TransactionKind {
self.transaction.kind()
}
/// Returns a measurement of the heap usage of this type and all its internals.
fn size(&self) -> usize {
self.transaction.transaction.input().len()
}
/// Returns the transaction type
fn tx_type(&self) -> u8 {
self.transaction.tx_type().into()
}
/// Returns the length of the rlp encoded object
fn encoded_length(&self) -> usize {
self.transaction.length()
}
/// Returns chain_id
fn chain_id(&self) -> Option<u64> {
self.transaction.chain_id()
}
}
impl EthPoolTransaction for EthPooledTransaction {
fn take_blob(&mut self) -> EthBlobTransactionSidecar {
if self.is_eip4844() {
std::mem::replace(&mut self.blob_sidecar, EthBlobTransactionSidecar::Missing)
} else {
EthBlobTransactionSidecar::None
}
}
fn as_eip4844(&self) -> Option<&TxEip4844> {
self.transaction.as_eip4844()
}
fn validate_blob(
&self,
sidecar: &BlobTransactionSidecar,
settings: &KzgSettings,
) -> Result<(), BlobTransactionValidationError> {
match &self.transaction.transaction {
Transaction::Eip4844(tx) => tx.validate_blob(sidecar, settings),
_ => Err(BlobTransactionValidationError::NotBlobTransaction(self.tx_type())),
}
}
}
impl FromRecoveredTransaction for EthPooledTransaction {
fn from_recovered_transaction(tx: TransactionSignedEcRecovered) -> Self {
EthPooledTransaction::new(tx)
}
}
impl FromRecoveredPooledTransaction for EthPooledTransaction {
fn from_recovered_transaction(tx: PooledTransactionsElementEcRecovered) -> Self {
EthPooledTransaction::from(tx)
}
}
impl IntoRecoveredTransaction for EthPooledTransaction {
fn to_recovered_transaction(&self) -> TransactionSignedEcRecovered {
self.transaction.clone()
}
}
/// Represents the current status of the pool.
#[derive(Debug, Clone, Default)]
pub struct PoolSize {
/// Number of transactions in the _pending_ sub-pool.
pub pending: usize,
/// Reported size of transactions in the _pending_ sub-pool.
pub pending_size: usize,
/// Number of transactions in the _basefee_ pool.
pub basefee: usize,
/// Reported size of transactions in the _basefee_ sub-pool.
pub basefee_size: usize,
/// Number of transactions in the _queued_ sub-pool.
pub queued: usize,
/// Reported size of transactions in the _queued_ sub-pool.
pub queued_size: usize,
/// Number of all transactions of all sub-pools
///
/// Note: this is the sum of ```pending + basefee + queued```
pub total: usize,
}
/// Represents the current status of the pool.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct BlockInfo {
/// Hash for the currently tracked block.
pub last_seen_block_hash: H256,
/// Current the currently tracked block.
pub last_seen_block_number: u64,
/// Currently enforced base fee: the threshold for the basefee sub-pool.
///
/// Note: this is the derived base fee of the _next_ block that builds on the clock the pool is
/// currently tracking.
pub pending_basefee: u64,
}
/// The limit to enforce for [TransactionPool::get_pooled_transaction_elements].
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum GetPooledTransactionLimit {
/// No limit, return all transactions.
None,
/// Enforce a size limit on the returned transactions, for example 2MB
SizeSoftLimit(usize),
}
impl GetPooledTransactionLimit {
/// Returns true if the given size exceeds the limit.
#[inline]
pub fn exceeds(&self, size: usize) -> bool {
match self {
GetPooledTransactionLimit::None => false,
GetPooledTransactionLimit::SizeSoftLimit(limit) => size > *limit,
}
}
}
/// A Stream that yields full transactions the subpool
#[must_use = "streams do nothing unless polled"]
#[derive(Debug)]
pub struct NewSubpoolTransactionStream<Tx: PoolTransaction> {
st: Receiver<NewTransactionEvent<Tx>>,
subpool: SubPool,
}
// === impl NewSubpoolTransactionStream ===
impl<Tx: PoolTransaction> NewSubpoolTransactionStream<Tx> {
/// Create a new stream that yields full transactions from the subpool
pub fn new(st: Receiver<NewTransactionEvent<Tx>>, subpool: SubPool) -> Self {
Self { st, subpool }
}
}
impl<Tx: PoolTransaction> Stream for NewSubpoolTransactionStream<Tx> {
type Item = NewTransactionEvent<Tx>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop {
match ready!(self.st.poll_recv(cx)) {
Some(event) => {
if event.subpool == self.subpool {
return Poll::Ready(Some(event))
}
}
None => return Poll::Ready(None),
}
}
}
}