anvil/eth/pool/
transactions.rs

1use crate::eth::{error::PoolError, util::hex_fmt_many};
2use alloy_consensus::{
3    Transaction, Typed2718,
4    crypto::RecoveryError,
5    transaction::{SignerRecoverable, TxHashRef},
6};
7use alloy_network::AnyRpcTransaction;
8use alloy_primitives::{
9    Address, TxHash,
10    map::{HashMap, HashSet},
11};
12use alloy_rlp::Encodable;
13use anvil_core::eth::transaction::PendingTransaction;
14use parking_lot::RwLock;
15use std::{
16    cmp::{Ordering, Reverse},
17    collections::{BTreeMap, BTreeSet},
18    fmt,
19    str::FromStr,
20    sync::Arc,
21    time::Instant,
22};
23
24/// A unique identifying marker for a transaction
25pub type TxMarker = Vec<u8>;
26
27/// Result type for replaced transactions: the replaced pool transactions and the hashes they
28/// unlock.
29type ReplacedTransactions<T> = (Vec<Arc<PoolTransaction<T>>>, Vec<TxHash>);
30
31/// Modes that determine the transaction ordering of the mempool
32///
33/// This type controls the transaction order via the priority metric of a transaction
34#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
35pub enum TransactionOrder {
36    /// Keep the pool transaction transactions sorted in the order they arrive.
37    ///
38    /// This will essentially assign every transaction the exact priority so the order is
39    /// determined by their internal id
40    Fifo,
41    /// This means that it prioritizes transactions based on the fees paid to the miner.
42    ///
43    /// Transactions are ranked by their effective miner tip at the base fee of the block being
44    /// built.
45    #[default]
46    Fees,
47}
48
49impl TransactionOrder {
50    /// Returns the priority of the transactions
51    pub fn priority<T: Transaction>(&self, tx: &T) -> TransactionPriority {
52        match self {
53            Self::Fifo => TransactionPriority::default(),
54            Self::Fees => TransactionPriority {
55                max_fee_per_gas: tx.max_fee_per_gas(),
56                max_priority_fee_per_gas: tx.max_priority_fee_per_gas(),
57            },
58        }
59    }
60}
61
62impl FromStr for TransactionOrder {
63    type Err = String;
64
65    fn from_str(s: &str) -> Result<Self, Self::Err> {
66        let s = s.to_lowercase();
67        let order = match s.as_str() {
68            "fees" => Self::Fees,
69            "fifo" => Self::Fifo,
70            _ => return Err(format!("Unknown TransactionOrder: `{s}`")),
71        };
72        Ok(order)
73    }
74}
75
76/// Metric value for the priority of a transaction.
77///
78/// The `TransactionPriority` determines the ordering of two transactions that have all their
79/// markers satisfied.
80#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
81pub struct TransactionPriority {
82    max_fee_per_gas: u128,
83    max_priority_fee_per_gas: Option<u128>,
84}
85
86impl TransactionPriority {
87    /// Returns the effective miner tip at the given base fee, see
88    /// [`Transaction::effective_tip_per_gas`].
89    fn effective_tip(&self, base_fee: u64) -> u128 {
90        let fee = self.max_fee_per_gas.saturating_sub(base_fee as u128);
91        self.max_priority_fee_per_gas.map_or(fee, |tip| fee.min(tip))
92    }
93}
94
95/// Internal Transaction type
96#[derive(Clone, PartialEq, Eq)]
97pub struct PoolTransaction<T> {
98    /// the pending eth transaction
99    pub pending_transaction: PendingTransaction<T>,
100    /// Markers required by the transaction
101    pub requires: Vec<TxMarker>,
102    /// Markers that this transaction provides
103    pub provides: Vec<TxMarker>,
104    /// priority of the transaction
105    pub priority: TransactionPriority,
106    /// Whether this transaction is being replayed from chain history.
107    pub is_replay: bool,
108}
109
110// == impl PoolTransaction ==
111
112impl<T> PoolTransaction<T> {
113    pub const fn new(transaction: PendingTransaction<T>) -> Self {
114        Self {
115            pending_transaction: transaction,
116            requires: vec![],
117            provides: vec![],
118            priority: TransactionPriority { max_fee_per_gas: 0, max_priority_fee_per_gas: None },
119            is_replay: false,
120        }
121    }
122
123    /// Marks this transaction as a historical replay.
124    pub const fn with_replay(mut self) -> Self {
125        self.is_replay = true;
126        self
127    }
128
129    /// Returns the hash of this transaction
130    pub const fn hash(&self) -> TxHash {
131        *self.pending_transaction.hash()
132    }
133}
134
135impl<T: Transaction> PoolTransaction<T> {
136    /// Returns the max fee per gas of this transaction
137    pub fn max_fee_per_gas(&self) -> u128 {
138        self.pending_transaction.transaction.max_fee_per_gas()
139    }
140}
141
142impl<T: Typed2718> PoolTransaction<T> {
143    /// Returns the type of the transaction
144    pub fn tx_type(&self) -> u8 {
145        self.pending_transaction.transaction.ty()
146    }
147}
148
149impl<T: fmt::Debug> fmt::Debug for PoolTransaction<T> {
150    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
151        write!(fmt, "Transaction {{ ")?;
152        write!(fmt, "hash: {:?}, ", self.pending_transaction.hash())?;
153        write!(fmt, "requires: [{}], ", hex_fmt_many(self.requires.iter()))?;
154        write!(fmt, "provides: [{}], ", hex_fmt_many(self.provides.iter()))?;
155        write!(fmt, "raw tx: {:?}", self.pending_transaction)?;
156        write!(fmt, "}}")?;
157        Ok(())
158    }
159}
160
161impl<T> TryFrom<AnyRpcTransaction> for PoolTransaction<T>
162where
163    T: SignerRecoverable + TxHashRef + Encodable + TryFrom<AnyRpcTransaction>,
164    <T as TryFrom<AnyRpcTransaction>>::Error: Into<eyre::Error>,
165    RecoveryError: Into<eyre::Error>,
166{
167    type Error = eyre::Error;
168    fn try_from(value: AnyRpcTransaction) -> Result<Self, Self::Error> {
169        let typed_transaction = T::try_from(value).map_err(Into::into)?;
170        let pending_transaction = PendingTransaction::new(typed_transaction)?;
171        Ok(Self {
172            pending_transaction,
173            requires: vec![],
174            provides: vec![],
175            priority: TransactionPriority::default(),
176            is_replay: false,
177        })
178    }
179}
180
181/// A waiting pool of transaction that are pending, but not yet ready to be included in a new block.
182///
183/// Keeps a set of transactions that are waiting for other transactions
184#[derive(Clone, Debug)]
185pub struct PendingTransactions<T> {
186    /// markers that aren't yet provided by any transaction
187    required_markers: HashMap<TxMarker, HashSet<TxHash>>,
188    /// mapping of the markers of a transaction to the hash of the transaction
189    waiting_markers: HashMap<Vec<TxMarker>, TxHash>,
190    /// the transactions that are not ready yet are waiting for another tx to finish
191    waiting_queue: HashMap<TxHash, PendingPoolTransaction<T>>,
192}
193
194impl<T> Default for PendingTransactions<T> {
195    fn default() -> Self {
196        Self {
197            required_markers: Default::default(),
198            waiting_markers: Default::default(),
199            waiting_queue: Default::default(),
200        }
201    }
202}
203
204impl<T> PendingTransactions<T> {
205    /// Returns an independent snapshot of the pending transactions.
206    pub(super) fn snapshot(&self) -> Self {
207        Self {
208            required_markers: self.required_markers.clone(),
209            waiting_markers: self.waiting_markers.clone(),
210            waiting_queue: self.waiting_queue.clone(),
211        }
212    }
213
214    /// Returns the number of transactions that are currently waiting
215    pub fn len(&self) -> usize {
216        self.waiting_queue.len()
217    }
218
219    pub fn is_empty(&self) -> bool {
220        self.waiting_queue.is_empty()
221    }
222
223    /// Clears internal state
224    pub fn clear(&mut self) {
225        self.required_markers.clear();
226        self.waiting_markers.clear();
227        self.waiting_queue.clear();
228    }
229
230    /// Returns an iterator over all transactions in the waiting pool
231    pub fn transactions(&self) -> impl Iterator<Item = Arc<PoolTransaction<T>>> + '_ {
232        self.waiting_queue.values().map(|tx| tx.transaction.clone())
233    }
234
235    /// Returns true if given transaction is part of the queue
236    pub fn contains(&self, hash: &TxHash) -> bool {
237        self.waiting_queue.contains_key(hash)
238    }
239
240    /// Returns the transaction for the hash if it's pending
241    pub fn get(&self, hash: &TxHash) -> Option<&PendingPoolTransaction<T>> {
242        self.waiting_queue.get(hash)
243    }
244
245    /// This will check off the markers of pending transactions.
246    ///
247    /// Returns the those transactions that become unlocked (all markers checked) and can be moved
248    /// to the ready queue.
249    pub fn mark_and_unlock(
250        &mut self,
251        markers: impl IntoIterator<Item = impl AsRef<TxMarker>>,
252    ) -> Vec<PendingPoolTransaction<T>> {
253        let mut unlocked_ready = Vec::new();
254        for mark in markers {
255            let mark = mark.as_ref();
256            if let Some(tx_hashes) = self.required_markers.remove(mark) {
257                for hash in tx_hashes {
258                    let tx = self.waiting_queue.get_mut(&hash).expect("tx is included;");
259                    tx.mark(mark);
260
261                    if tx.is_ready() {
262                        let tx = self.waiting_queue.remove(&hash).expect("tx is included;");
263                        self.waiting_markers.remove(&tx.transaction.provides);
264
265                        unlocked_ready.push(tx);
266                    }
267                }
268            }
269        }
270
271        unlocked_ready
272    }
273
274    /// Removes the transactions associated with the given hashes
275    ///
276    /// Returns all removed transactions.
277    pub fn remove(&mut self, hashes: Vec<TxHash>) -> Vec<Arc<PoolTransaction<T>>> {
278        let mut removed = vec![];
279        for hash in hashes {
280            if let Some(waiting_tx) = self.waiting_queue.remove(&hash) {
281                self.waiting_markers.remove(&waiting_tx.transaction.provides);
282                for marker in waiting_tx.missing_markers {
283                    let remove = if let Some(required) = self.required_markers.get_mut(&marker) {
284                        required.remove(&hash);
285                        required.is_empty()
286                    } else {
287                        false
288                    };
289                    if remove {
290                        self.required_markers.remove(&marker);
291                    }
292                }
293                removed.push(waiting_tx.transaction)
294            }
295        }
296        removed
297    }
298
299    /// Removes transactions and their transitive dependents from the waiting pool.
300    pub fn remove_with_dependents(
301        &mut self,
302        hashes: Vec<TxHash>,
303        invalidated: impl IntoIterator<Item = TxMarker>,
304    ) -> Vec<Arc<PoolTransaction<T>>> {
305        let mut required_by = HashMap::<TxMarker, Vec<TxHash>>::default();
306        for (hash, tx) in &self.waiting_queue {
307            for marker in &tx.transaction.requires {
308                required_by.entry(marker.clone()).or_default().push(*hash);
309            }
310        }
311
312        let mut to_remove = HashSet::<TxHash>::default();
313        let mut markers = invalidated.into_iter().collect::<Vec<_>>();
314        for hash in hashes {
315            if to_remove.insert(hash)
316                && let Some(tx) = self.waiting_queue.get(&hash)
317            {
318                markers.extend(tx.transaction.provides.iter().cloned());
319            }
320        }
321
322        while let Some(marker) = markers.pop() {
323            if let Some(dependents) = required_by.remove(&marker) {
324                for hash in dependents {
325                    if to_remove.insert(hash)
326                        && let Some(tx) = self.waiting_queue.get(&hash)
327                    {
328                        markers.extend(tx.transaction.provides.iter().cloned());
329                    }
330                }
331            }
332        }
333
334        self.remove(to_remove.into_iter().collect())
335    }
336}
337
338impl<T: Transaction> PendingTransactions<T> {
339    /// Adds a transaction to Pending queue of transactions
340    pub fn add_transaction(&mut self, tx: PendingPoolTransaction<T>) -> Result<(), PoolError> {
341        assert!(!tx.is_ready(), "transaction must not be ready");
342        assert!(
343            !self.waiting_queue.contains_key(&tx.transaction.hash()),
344            "transaction is already added"
345        );
346
347        let replaced_hash = if let Some(replace) = self
348            .waiting_markers
349            .get(&tx.transaction.provides)
350            .and_then(|hash| self.waiting_queue.get(hash))
351        {
352            // check if underpriced
353            if tx.transaction.max_fee_per_gas() <= replace.transaction.max_fee_per_gas() {
354                warn!(target: "txpool", "pending replacement transaction underpriced [{:?}]", tx.transaction.hash());
355                return Err(PoolError::ReplacementUnderpriced(tx.transaction.hash()));
356            }
357            Some(replace.transaction.hash())
358        } else {
359            None
360        };
361        // Remove old markers before inserting the replacement, which shares their keys.
362        if let Some(replaced_hash) = replaced_hash {
363            self.remove(vec![replaced_hash]);
364        }
365
366        // add all missing markers
367        for marker in &tx.missing_markers {
368            self.required_markers.entry(marker.clone()).or_default().insert(tx.transaction.hash());
369        }
370
371        // also track identifying markers
372        self.waiting_markers.insert(tx.transaction.provides.clone(), tx.transaction.hash());
373        // add tx to the queue
374        self.waiting_queue.insert(tx.transaction.hash(), tx);
375
376        Ok(())
377    }
378}
379
380/// A transaction in the pool
381pub struct PendingPoolTransaction<T> {
382    pub transaction: Arc<PoolTransaction<T>>,
383    /// markers required and have not been satisfied yet by other transactions in the pool
384    pub missing_markers: HashSet<TxMarker>,
385    /// timestamp when the tx was added
386    pub added_at: Instant,
387}
388
389impl<T> Clone for PendingPoolTransaction<T> {
390    fn clone(&self) -> Self {
391        Self {
392            transaction: Arc::clone(&self.transaction),
393            missing_markers: self.missing_markers.clone(),
394            added_at: self.added_at,
395        }
396    }
397}
398
399impl<T> PendingPoolTransaction<T> {
400    /// Creates a new `PendingPoolTransaction`.
401    ///
402    /// Determines the markers that are still missing before this transaction can be moved to the
403    /// ready queue.
404    pub fn new(transaction: PoolTransaction<T>, provided: &HashMap<TxMarker, TxHash>) -> Self {
405        let missing_markers = transaction
406            .requires
407            .iter()
408            .filter(|marker| {
409                // is true if the marker is already satisfied either via transaction in the pool
410                !provided.contains_key(&**marker)
411            })
412            .cloned()
413            .collect();
414
415        Self { transaction: Arc::new(transaction), missing_markers, added_at: Instant::now() }
416    }
417
418    /// Removes the required marker
419    pub fn mark(&mut self, marker: &TxMarker) {
420        self.missing_markers.remove(marker);
421    }
422
423    /// Returns true if transaction has all requirements satisfied.
424    pub fn is_ready(&self) -> bool {
425        self.missing_markers.is_empty()
426    }
427}
428
429impl<T: fmt::Debug> fmt::Debug for PendingPoolTransaction<T> {
430    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
431        write!(fmt, "PendingTransaction {{ ")?;
432        write!(fmt, "added_at: {:?}, ", self.added_at)?;
433        write!(fmt, "tx: {:?}, ", self.transaction)?;
434        write!(fmt, "missing_markers: {{{}}}", hex_fmt_many(self.missing_markers.iter()))?;
435        write!(fmt, "}}")
436    }
437}
438
439pub struct TransactionsIterator<T> {
440    all: HashMap<TxHash, ReadyTransaction<T>>,
441    awaiting: HashMap<TxHash, (usize, PoolTransactionRef<T>)>,
442    /// Independent transactions keyed by their effective tip at `base_fee`, then arrival order.
443    independent: BTreeMap<(u128, Reverse<u64>), PoolTransactionRef<T>>,
444    base_fee: u64,
445    _invalid: HashSet<TxHash>,
446}
447
448impl<T> TransactionsIterator<T> {
449    /// Adds an independent transaction ranked at the iterator's base fee.
450    fn insert_independent(&mut self, tx_ref: PoolTransactionRef<T>) {
451        let tip = tx_ref.transaction.priority.effective_tip(self.base_fee);
452        self.independent.insert((tip, Reverse(tx_ref.id)), tx_ref);
453    }
454
455    /// Depending on number of satisfied requirements insert given ref
456    /// either to awaiting set or to best set.
457    fn independent_or_awaiting(&mut self, satisfied: usize, tx_ref: PoolTransactionRef<T>) {
458        if satisfied >= tx_ref.transaction.requires.len() {
459            // If we have satisfied all deps insert to best
460            self.insert_independent(tx_ref);
461        } else {
462            // otherwise we're still awaiting for some deps
463            self.awaiting.insert(tx_ref.transaction.hash(), (satisfied, tx_ref));
464        }
465    }
466}
467
468impl<T> Iterator for TransactionsIterator<T> {
469    type Item = Arc<PoolTransaction<T>>;
470
471    fn next(&mut self) -> Option<Self::Item> {
472        loop {
473            let (_, best) = self.independent.pop_last()?;
474            let hash = best.transaction.hash();
475
476            let ready =
477                if let Some(ready) = self.all.get(&hash).cloned() { ready } else { continue };
478
479            // Insert transactions that just got unlocked.
480            for hash in &ready.unlocks {
481                // first check local awaiting transactions
482                let res = if let Some((mut satisfied, tx_ref)) = self.awaiting.remove(hash) {
483                    satisfied += 1;
484                    Some((satisfied, tx_ref))
485                    // then get from the pool
486                } else {
487                    self.all
488                        .get(hash)
489                        .map(|next| (next.requires_offset + 1, next.transaction.clone()))
490                };
491                if let Some((satisfied, tx_ref)) = res {
492                    self.independent_or_awaiting(satisfied, tx_ref)
493                }
494            }
495
496            return Some(best.transaction);
497        }
498    }
499}
500
501/// transactions that are ready to be included in a block.
502#[derive(Clone, Debug)]
503pub struct ReadyTransactions<T> {
504    /// keeps track of transactions inserted in the pool
505    ///
506    /// this way we can determine when transactions where submitted to the pool
507    id: u64,
508    /// markers that are provided by `ReadyTransaction`s
509    provided_markers: HashMap<TxMarker, TxHash>,
510    /// transactions that are ready
511    ready_tx: Arc<RwLock<HashMap<TxHash, ReadyTransaction<T>>>>,
512    /// independent transactions that can be included directly and don't require other transactions
513    /// Sorted by their id
514    independent_transactions: BTreeSet<PoolTransactionRef<T>>,
515}
516
517impl<T> Default for ReadyTransactions<T> {
518    fn default() -> Self {
519        Self {
520            id: 0,
521            provided_markers: Default::default(),
522            ready_tx: Default::default(),
523            independent_transactions: Default::default(),
524        }
525    }
526}
527
528impl<T> ReadyTransactions<T> {
529    /// Returns an independent snapshot of the ready transactions.
530    pub(super) fn snapshot(&self) -> Self {
531        Self {
532            id: self.id,
533            provided_markers: self.provided_markers.clone(),
534            ready_tx: Arc::new(RwLock::new(self.ready_tx.read().clone())),
535            independent_transactions: self.independent_transactions.clone(),
536        }
537    }
538
539    /// Returns an iterator over all transactions, ranked at the given base fee.
540    pub fn get_transactions(&self, base_fee: u64) -> TransactionsIterator<T> {
541        let mut iter = TransactionsIterator {
542            all: self.ready_tx.read().clone(),
543            independent: Default::default(),
544            base_fee,
545            awaiting: Default::default(),
546            _invalid: Default::default(),
547        };
548        for tx_ref in &self.independent_transactions {
549            iter.insert_independent(tx_ref.clone());
550        }
551        iter
552    }
553
554    /// Clears the internal state
555    pub fn clear(&mut self) {
556        self.provided_markers.clear();
557        self.ready_tx.write().clear();
558        self.independent_transactions.clear();
559    }
560
561    /// Returns true if the transaction is part of the queue.
562    pub fn contains(&self, hash: &TxHash) -> bool {
563        self.ready_tx.read().contains_key(hash)
564    }
565
566    /// Returns the number of ready transactions without cloning the snapshot
567    pub fn len(&self) -> usize {
568        self.ready_tx.read().len()
569    }
570
571    /// Returns true if there are no ready transactions
572    pub fn is_empty(&self) -> bool {
573        self.ready_tx.read().is_empty()
574    }
575
576    /// Returns the transaction for the hash if it's in the ready pool but not yet mined
577    pub fn get(&self, hash: &TxHash) -> Option<ReadyTransaction<T>> {
578        self.ready_tx.read().get(hash).cloned()
579    }
580
581    pub const fn provided_markers(&self) -> &HashMap<TxMarker, TxHash> {
582        &self.provided_markers
583    }
584
585    const fn next_id(&mut self) -> u64 {
586        let id = self.id;
587        self.id = self.id.wrapping_add(1);
588        id
589    }
590
591    /// Removes the transactions from the ready queue and returns the removed transactions.
592    /// This will also remove all transactions that depend on those.
593    pub fn clear_transactions(&mut self, tx_hashes: &[TxHash]) -> Vec<Arc<PoolTransaction<T>>> {
594        self.remove_with_markers(tx_hashes.to_vec(), None)
595    }
596
597    /// Removes the transactions that provide the marker
598    ///
599    /// This will also remove all transactions that lead to the transaction that provides the
600    /// marker.
601    pub fn prune_tags(&mut self, marker: TxMarker) -> Vec<Arc<PoolTransaction<T>>> {
602        let mut removed_tx = vec![];
603
604        // the markers to remove
605        let mut remove = vec![marker];
606
607        while let Some(marker) = remove.pop() {
608            let res = self
609                .provided_markers
610                .remove(&marker)
611                .and_then(|hash| self.ready_tx.write().remove(&hash));
612
613            if let Some(tx) = res {
614                let unlocks = tx.unlocks;
615                self.independent_transactions.remove(&tx.transaction);
616                let tx = tx.transaction.transaction;
617
618                // also prune previous transactions
619                {
620                    let hash = tx.hash();
621                    let mut ready = self.ready_tx.write();
622
623                    let mut previous_markers = |marker| -> Option<Vec<TxMarker>> {
624                        let prev_hash = self.provided_markers.get(marker)?;
625                        let tx2 = ready.get_mut(prev_hash)?;
626                        // remove hash
627                        if let Some(idx) = tx2.unlocks.iter().position(|i| i == &hash) {
628                            tx2.unlocks.swap_remove(idx);
629                        }
630                        tx2.unlocks.is_empty().then(|| tx2.transaction.transaction.provides.clone())
631                    };
632
633                    // find previous transactions
634                    for marker in &tx.requires {
635                        if let Some(mut tags_to_remove) = previous_markers(marker) {
636                            remove.append(&mut tags_to_remove);
637                        }
638                    }
639                }
640
641                // add the transactions that just got unlocked to independent set
642                for hash in unlocks {
643                    if let Some(tx) = self.ready_tx.write().get_mut(&hash) {
644                        tx.requires_offset += 1;
645                        if tx.requires_offset == tx.transaction.transaction.requires.len() {
646                            self.independent_transactions.insert(tx.transaction.clone());
647                        }
648                    }
649                }
650                // finally, remove the markers that this transaction provides
651                let current_marker = &marker;
652                for marker in &tx.provides {
653                    let removed = self.provided_markers.remove(marker);
654                    assert_eq!(
655                        removed,
656                        if current_marker == marker { None } else { Some(tx.hash()) },
657                        "The pool contains exactly one transaction providing given tag; the removed transaction
658						claims to provide that tag, so it has to be mapped to it's hash; qed"
659                    );
660                }
661                removed_tx.push(tx);
662            }
663        }
664
665        removed_tx
666    }
667
668    /// Removes transactions and those that depend on them and satisfy at least one marker in the
669    /// given filter set.
670    pub fn remove_with_markers(
671        &mut self,
672        mut tx_hashes: Vec<TxHash>,
673        marker_filter: Option<HashSet<TxMarker>>,
674    ) -> Vec<Arc<PoolTransaction<T>>> {
675        let mut removed = Vec::new();
676        let mut ready = self.ready_tx.write();
677
678        while let Some(hash) = tx_hashes.pop() {
679            if let Some(mut tx) = ready.remove(&hash) {
680                let invalidated = tx.transaction.transaction.provides.iter().filter(|mark| {
681                    marker_filter.as_ref().is_none_or(|filter| !filter.contains(&**mark))
682                });
683
684                let mut removed_some_marks = false;
685                // remove entries from provided_markers
686                for mark in invalidated {
687                    removed_some_marks = true;
688                    self.provided_markers.remove(mark);
689                }
690
691                // remove from unlocks
692                for mark in &tx.transaction.transaction.requires {
693                    if let Some(provider_hash) = self.provided_markers.get(mark)
694                        && let Some(provider_tx) = ready.get_mut(provider_hash)
695                        && let Some(idx) = provider_tx.unlocks.iter().position(|i| i == &hash)
696                    {
697                        provider_tx.unlocks.swap_remove(idx);
698                    }
699                }
700
701                // remove from the independent set
702                self.independent_transactions.remove(&tx.transaction);
703
704                if removed_some_marks {
705                    // remove all transactions that the current one unlocks
706                    tx_hashes.append(&mut tx.unlocks);
707                }
708
709                // remove transaction
710                removed.push(tx.transaction.transaction);
711            }
712        }
713
714        removed
715    }
716}
717
718impl<T: Transaction> ReadyTransactions<T> {
719    /// Adds a new transactions to the ready queue.
720    ///
721    /// # Panics
722    ///
723    /// If the pending transaction is not ready ([`PendingPoolTransaction::is_ready`])
724    /// or the transaction is already included.
725    pub fn add_transaction(
726        &mut self,
727        tx: PendingPoolTransaction<T>,
728    ) -> Result<Vec<Arc<PoolTransaction<T>>>, PoolError> {
729        assert!(tx.is_ready(), "transaction must be ready",);
730        assert!(
731            !self.ready_tx.read().contains_key(&tx.transaction.hash()),
732            "transaction already included"
733        );
734
735        let (replaced_tx, unlocks) = self.replaced_transactions(&tx.transaction)?;
736
737        let id = self.next_id();
738        let hash = tx.transaction.hash();
739
740        let mut independent = true;
741        let mut requires_offset = 0;
742        let mut ready = self.ready_tx.write();
743        // Add links to transactions that unlock the current one
744        for mark in &tx.transaction.requires {
745            // Check if the transaction that satisfies the mark is still in the queue.
746            if let Some(other) = self.provided_markers.get(mark) {
747                let tx = ready.get_mut(other).expect("hash included;");
748                tx.unlocks.push(hash);
749                // tx still depends on other tx
750                independent = false;
751            } else {
752                requires_offset += 1;
753            }
754        }
755
756        // update markers
757        for mark in tx.transaction.provides.iter().cloned() {
758            self.provided_markers.insert(mark, hash);
759        }
760
761        let transaction = PoolTransactionRef { id, transaction: tx.transaction };
762
763        // add to the independent set
764        if independent {
765            self.independent_transactions.insert(transaction.clone());
766        }
767
768        // insert to ready queue
769        ready.insert(hash, ReadyTransaction { transaction, unlocks, requires_offset });
770
771        Ok(replaced_tx)
772    }
773
774    /// Removes and returns those transactions that got replaced by the `tx`
775    fn replaced_transactions(
776        &mut self,
777        tx: &PoolTransaction<T>,
778    ) -> Result<ReplacedTransactions<T>, PoolError> {
779        // check if we are replacing transactions
780        let remove_hashes: HashSet<_> =
781            tx.provides.iter().filter_map(|mark| self.provided_markers.get(mark)).collect();
782
783        // early exit if we are not replacing anything.
784        if remove_hashes.is_empty() {
785            return Ok((Vec::new(), Vec::new()));
786        }
787
788        // check if we're replacing the same transaction and if it can be replaced
789        let mut unlocked_tx = Vec::new();
790        {
791            // construct a list of unlocked transactions
792            // also check for transactions that shouldn't be replaced because underpriced
793            let ready = self.ready_tx.read();
794            for to_remove in remove_hashes.iter().filter_map(|hash| ready.get(*hash)) {
795                // if we're attempting to replace a transaction that provides the exact same markers
796                // (addr + nonce) then we check for gas price
797                if to_remove.provides() == tx.provides {
798                    // check if underpriced
799                    if tx.pending_transaction.transaction.max_fee_per_gas()
800                        <= to_remove.max_fee_per_gas()
801                    {
802                        warn!(target: "txpool", "ready replacement transaction underpriced [{:?}]", tx.hash());
803                        return Err(PoolError::ReplacementUnderpriced(tx.hash()));
804                    }
805                    trace!(target: "txpool", "replacing ready transaction [{:?}] with higher gas price [{:?}]", to_remove.transaction.transaction.hash(), tx.hash());
806                }
807
808                unlocked_tx.extend(to_remove.unlocks.iter().copied())
809            }
810        }
811
812        let remove_hashes = remove_hashes.into_iter().copied().collect::<Vec<_>>();
813
814        let new_provides = tx.provides.iter().cloned().collect::<HashSet<_>>();
815        let removed_tx = self.remove_with_markers(remove_hashes, Some(new_provides));
816
817        Ok((removed_tx, unlocked_tx))
818    }
819}
820
821/// A reference to a transaction in the pool
822#[derive(Debug)]
823pub struct PoolTransactionRef<T> {
824    /// actual transaction
825    pub transaction: Arc<PoolTransaction<T>>,
826    /// identifier used to internally compare the transaction in the pool
827    pub id: u64,
828}
829
830impl<T> Clone for PoolTransactionRef<T> {
831    fn clone(&self) -> Self {
832        Self { transaction: Arc::clone(&self.transaction), id: self.id }
833    }
834}
835
836impl<T> Eq for PoolTransactionRef<T> {}
837
838impl<T> PartialEq<Self> for PoolTransactionRef<T> {
839    fn eq(&self, other: &Self) -> bool {
840        self.cmp(other) == Ordering::Equal
841    }
842}
843
844impl<T> PartialOrd<Self> for PoolTransactionRef<T> {
845    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
846        Some(self.cmp(other))
847    }
848}
849
850impl<T> Ord for PoolTransactionRef<T> {
851    fn cmp(&self, other: &Self) -> Ordering {
852        self.transaction
853            .priority
854            .cmp(&other.transaction.priority)
855            .then_with(|| other.id.cmp(&self.id))
856    }
857}
858
859#[derive(Debug)]
860pub struct ReadyTransaction<T> {
861    /// ref to the actual transaction
862    pub transaction: PoolTransactionRef<T>,
863    /// tracks the transactions that get unlocked by this transaction
864    pub unlocks: Vec<TxHash>,
865    /// amount of required markers that are inherently provided
866    pub requires_offset: usize,
867}
868
869impl<T> Clone for ReadyTransaction<T> {
870    fn clone(&self) -> Self {
871        Self {
872            transaction: self.transaction.clone(),
873            unlocks: self.unlocks.clone(),
874            requires_offset: self.requires_offset,
875        }
876    }
877}
878
879impl<T> ReadyTransaction<T> {
880    pub fn provides(&self) -> &[TxMarker] {
881        &self.transaction.transaction.provides
882    }
883}
884
885impl<T: Transaction> ReadyTransaction<T> {
886    pub fn max_fee_per_gas(&self) -> u128 {
887        self.transaction.transaction.max_fee_per_gas()
888    }
889}
890
891/// creates an unique identifier for aan (`nonce` + `Address`) combo
892pub fn to_marker(nonce: u64, from: Address) -> TxMarker {
893    let mut data = [0u8; 28];
894    data[..8].copy_from_slice(&nonce.to_le_bytes()[..]);
895    data[8..].copy_from_slice(&from.0[..]);
896    data.to_vec()
897}
898
899#[cfg(test)]
900mod tests {
901    use super::*;
902
903    #[test]
904    fn can_id_txs() {
905        let addr = Address::random();
906        assert_eq!(to_marker(1, addr), to_marker(1, addr));
907        assert_ne!(to_marker(2, addr), to_marker(1, addr));
908    }
909}
anvil/eth/pool/transactions.rs

anvil/eth/pool/
transactions.rs
