reth_eth_wire_types/
broadcast.rs

1//! Types for broadcasting new data.
2
3use crate::{EthMessage, EthVersion, NetworkPrimitives};
4use alloc::{sync::Arc, vec::Vec};
5use alloy_primitives::{
6    map::{HashMap, HashSet},
7    Bytes, TxHash, B256, U128,
8};
9use alloy_rlp::{
10    Decodable, Encodable, RlpDecodable, RlpDecodableWrapper, RlpEncodable, RlpEncodableWrapper,
11};
12use core::mem;
13use derive_more::{Constructor, Deref, DerefMut, From, IntoIterator};
14use reth_codecs_derive::{add_arbitrary_tests, generate_tests};
15use reth_ethereum_primitives::TransactionSigned;
16use reth_primitives_traits::SignedTransaction;
17
18/// This informs peers of new blocks that have appeared on the network.
19#[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Default)]
20#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
21#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
22#[add_arbitrary_tests(rlp)]
23pub struct NewBlockHashes(
24    /// New block hashes and the block number for each blockhash.
25    /// Clients should request blocks using a [`GetBlockBodies`](crate::GetBlockBodies) message.
26    pub Vec<BlockHashNumber>,
27);
28
29// === impl NewBlockHashes ===
30
31impl NewBlockHashes {
32    /// Returns the latest block in the list of blocks.
33    pub fn latest(&self) -> Option<&BlockHashNumber> {
34        self.0.iter().fold(None, |latest, block| {
35            if let Some(latest) = latest {
36                return if latest.number > block.number { Some(latest) } else { Some(block) }
37            }
38            Some(block)
39        })
40    }
41}
42
43/// A block hash _and_ a block number.
44#[derive(Clone, Debug, PartialEq, Eq, RlpEncodable, RlpDecodable, Default)]
45#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
46#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
47#[add_arbitrary_tests(rlp)]
48pub struct BlockHashNumber {
49    /// The block hash
50    pub hash: B256,
51    /// The block number
52    pub number: u64,
53}
54
55impl From<Vec<BlockHashNumber>> for NewBlockHashes {
56    fn from(v: Vec<BlockHashNumber>) -> Self {
57        Self(v)
58    }
59}
60
61impl From<NewBlockHashes> for Vec<BlockHashNumber> {
62    fn from(v: NewBlockHashes) -> Self {
63        v.0
64    }
65}
66
67/// A new block with the current total difficulty, which includes the difficulty of the returned
68/// block.
69#[derive(Clone, Debug, PartialEq, Eq, RlpEncodable, RlpDecodable, Default)]
70#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
71#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
72pub struct NewBlock<B = reth_ethereum_primitives::Block> {
73    /// A new block.
74    pub block: B,
75    /// The current total difficulty.
76    pub td: U128,
77}
78
79generate_tests!(#[rlp, 25] NewBlock<reth_ethereum_primitives::Block>, EthNewBlockTests);
80
81/// This informs peers of transactions that have appeared on the network and are not yet included
82/// in a block.
83#[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Default)]
84#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
85#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
86#[add_arbitrary_tests(rlp, 10)]
87pub struct Transactions<T = TransactionSigned>(
88    /// New transactions for the peer to include in its mempool.
89    pub Vec<T>,
90);
91
92impl<T: SignedTransaction> Transactions<T> {
93    /// Returns `true` if the list of transactions contains any blob transactions.
94    pub fn has_eip4844(&self) -> bool {
95        self.0.iter().any(|tx| tx.is_eip4844())
96    }
97}
98
99impl<T> From<Vec<T>> for Transactions<T> {
100    fn from(txs: Vec<T>) -> Self {
101        Self(txs)
102    }
103}
104
105impl<T> From<Transactions<T>> for Vec<T> {
106    fn from(txs: Transactions<T>) -> Self {
107        txs.0
108    }
109}
110
111/// Same as [`Transactions`] but this is intended as egress message send from local to _many_ peers.
112///
113/// The list of transactions is constructed on per-peers basis, but the underlying transaction
114/// objects are shared.
115#[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper)]
116#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
117#[add_arbitrary_tests(rlp, 20)]
118pub struct SharedTransactions<T = TransactionSigned>(
119    /// New transactions for the peer to include in its mempool.
120    pub Vec<Arc<T>>,
121);
122
123/// A wrapper type for all different new pooled transaction types
124#[derive(Clone, Debug, PartialEq, Eq)]
125#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
126pub enum NewPooledTransactionHashes {
127    /// A list of transaction hashes valid for [66-68)
128    Eth66(NewPooledTransactionHashes66),
129    /// A list of transaction hashes valid from [68..]
130    ///
131    /// Note: it is assumed that the payload is valid (all vectors have the same length)
132    Eth68(NewPooledTransactionHashes68),
133}
134
135// === impl NewPooledTransactionHashes ===
136
137impl NewPooledTransactionHashes {
138    /// Returns the message [`EthVersion`].
139    pub const fn version(&self) -> EthVersion {
140        match self {
141            Self::Eth66(_) => EthVersion::Eth66,
142            Self::Eth68(_) => EthVersion::Eth68,
143        }
144    }
145
146    /// Returns `true` if the payload is valid for the given version
147    pub const fn is_valid_for_version(&self, version: EthVersion) -> bool {
148        match self {
149            Self::Eth66(_) => {
150                matches!(version, EthVersion::Eth67 | EthVersion::Eth66)
151            }
152            Self::Eth68(_) => {
153                matches!(version, EthVersion::Eth68)
154            }
155        }
156    }
157
158    /// Returns an iterator over all transaction hashes.
159    pub fn iter_hashes(&self) -> impl Iterator<Item = &B256> + '_ {
160        match self {
161            Self::Eth66(msg) => msg.0.iter(),
162            Self::Eth68(msg) => msg.hashes.iter(),
163        }
164    }
165
166    /// Returns an immutable reference to transaction hashes.
167    pub const fn hashes(&self) -> &Vec<B256> {
168        match self {
169            Self::Eth66(msg) => &msg.0,
170            Self::Eth68(msg) => &msg.hashes,
171        }
172    }
173
174    /// Returns a mutable reference to transaction hashes.
175    pub const fn hashes_mut(&mut self) -> &mut Vec<B256> {
176        match self {
177            Self::Eth66(msg) => &mut msg.0,
178            Self::Eth68(msg) => &mut msg.hashes,
179        }
180    }
181
182    /// Consumes the type and returns all hashes
183    pub fn into_hashes(self) -> Vec<B256> {
184        match self {
185            Self::Eth66(msg) => msg.0,
186            Self::Eth68(msg) => msg.hashes,
187        }
188    }
189
190    /// Returns an iterator over all transaction hashes.
191    pub fn into_iter_hashes(self) -> impl Iterator<Item = B256> {
192        match self {
193            Self::Eth66(msg) => msg.0.into_iter(),
194            Self::Eth68(msg) => msg.hashes.into_iter(),
195        }
196    }
197
198    /// Shortens the number of hashes in the message, keeping the first `len` hashes and dropping
199    /// the rest. If `len` is greater than the number of hashes, this has no effect.
200    pub fn truncate(&mut self, len: usize) {
201        match self {
202            Self::Eth66(msg) => msg.0.truncate(len),
203            Self::Eth68(msg) => {
204                msg.types.truncate(len);
205                msg.sizes.truncate(len);
206                msg.hashes.truncate(len);
207            }
208        }
209    }
210
211    /// Returns true if the message is empty
212    pub fn is_empty(&self) -> bool {
213        match self {
214            Self::Eth66(msg) => msg.0.is_empty(),
215            Self::Eth68(msg) => msg.hashes.is_empty(),
216        }
217    }
218
219    /// Returns the number of hashes in the message
220    pub fn len(&self) -> usize {
221        match self {
222            Self::Eth66(msg) => msg.0.len(),
223            Self::Eth68(msg) => msg.hashes.len(),
224        }
225    }
226
227    /// Returns an immutable reference to the inner type if this an eth68 announcement.
228    pub const fn as_eth68(&self) -> Option<&NewPooledTransactionHashes68> {
229        match self {
230            Self::Eth66(_) => None,
231            Self::Eth68(msg) => Some(msg),
232        }
233    }
234
235    /// Returns a mutable reference to the inner type if this an eth68 announcement.
236    pub const fn as_eth68_mut(&mut self) -> Option<&mut NewPooledTransactionHashes68> {
237        match self {
238            Self::Eth66(_) => None,
239            Self::Eth68(msg) => Some(msg),
240        }
241    }
242
243    /// Returns a mutable reference to the inner type if this an eth66 announcement.
244    pub const fn as_eth66_mut(&mut self) -> Option<&mut NewPooledTransactionHashes66> {
245        match self {
246            Self::Eth66(msg) => Some(msg),
247            Self::Eth68(_) => None,
248        }
249    }
250
251    /// Returns the inner type if this an eth68 announcement.
252    pub fn take_eth68(&mut self) -> Option<NewPooledTransactionHashes68> {
253        match self {
254            Self::Eth66(_) => None,
255            Self::Eth68(msg) => Some(mem::take(msg)),
256        }
257    }
258
259    /// Returns the inner type if this an eth66 announcement.
260    pub fn take_eth66(&mut self) -> Option<NewPooledTransactionHashes66> {
261        match self {
262            Self::Eth66(msg) => Some(mem::take(msg)),
263            Self::Eth68(_) => None,
264        }
265    }
266}
267
268impl<N: NetworkPrimitives> From<NewPooledTransactionHashes> for EthMessage<N> {
269    fn from(value: NewPooledTransactionHashes) -> Self {
270        match value {
271            NewPooledTransactionHashes::Eth66(msg) => Self::NewPooledTransactionHashes66(msg),
272            NewPooledTransactionHashes::Eth68(msg) => Self::NewPooledTransactionHashes68(msg),
273        }
274    }
275}
276
277impl From<NewPooledTransactionHashes66> for NewPooledTransactionHashes {
278    fn from(hashes: NewPooledTransactionHashes66) -> Self {
279        Self::Eth66(hashes)
280    }
281}
282
283impl From<NewPooledTransactionHashes68> for NewPooledTransactionHashes {
284    fn from(hashes: NewPooledTransactionHashes68) -> Self {
285        Self::Eth68(hashes)
286    }
287}
288
289/// This informs peers of transaction hashes for transactions that have appeared on the network,
290/// but have not been included in a block.
291#[derive(Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Default)]
292#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
293#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
294#[add_arbitrary_tests(rlp)]
295pub struct NewPooledTransactionHashes66(
296    /// Transaction hashes for new transactions that have appeared on the network.
297    /// Clients should request the transactions with the given hashes using a
298    /// [`GetPooledTransactions`](crate::GetPooledTransactions) message.
299    pub Vec<B256>,
300);
301
302impl From<Vec<B256>> for NewPooledTransactionHashes66 {
303    fn from(v: Vec<B256>) -> Self {
304        Self(v)
305    }
306}
307
308/// Same as [`NewPooledTransactionHashes66`] but extends that beside the transaction hashes,
309/// the node sends the transaction types and their sizes (as defined in EIP-2718) as well.
310#[derive(Clone, Debug, PartialEq, Eq, Default)]
311#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
312pub struct NewPooledTransactionHashes68 {
313    /// Transaction types for new transactions that have appeared on the network.
314    ///
315    /// ## Note on RLP encoding and decoding
316    ///
317    /// In the [eth/68 spec](https://eips.ethereum.org/EIPS/eip-5793#specification) this is defined
318    /// the following way:
319    ///  * `[type_0: B_1, type_1: B_1, ...]`
320    ///
321    /// This would make it seem like the [`Encodable`] and
322    /// [`Decodable`] implementations should directly use a `Vec<u8>` for
323    /// encoding and decoding, because it looks like this field should be encoded as a _list_ of
324    /// bytes.
325    ///
326    /// However, [this is implemented in geth as a `[]byte`
327    /// type](https://github.com/ethereum/go-ethereum/blob/82d934b1dd80cdd8190803ea9f73ed2c345e2576/eth/protocols/eth/protocol.go#L308-L313),
328    /// which [ends up being encoded as a RLP
329    /// string](https://github.com/ethereum/go-ethereum/blob/82d934b1dd80cdd8190803ea9f73ed2c345e2576/rlp/encode_test.go#L171-L176),
330    /// **not** a RLP list.
331    ///
332    /// Because of this, we do not directly use the `Vec<u8>` when encoding and decoding, and
333    /// instead use the [`Encodable`] and [`Decodable`]
334    /// implementations for `&[u8]` instead, which encodes into a RLP string, and expects an RLP
335    /// string when decoding.
336    pub types: Vec<u8>,
337    /// Transaction sizes for new transactions that have appeared on the network.
338    pub sizes: Vec<usize>,
339    /// Transaction hashes for new transactions that have appeared on the network.
340    pub hashes: Vec<B256>,
341}
342
343#[cfg(feature = "arbitrary")]
344impl proptest::prelude::Arbitrary for NewPooledTransactionHashes68 {
345    type Parameters = ();
346    fn arbitrary_with(_args: ()) -> Self::Strategy {
347        use proptest::{collection::vec, prelude::*};
348        // Generate a single random length for all vectors
349        let vec_length = any::<usize>().prop_map(|x| x % 100 + 1); // Lengths between 1 and 100
350
351        vec_length
352            .prop_flat_map(|len| {
353                // Use the generated length to create vectors of TxType, usize, and B256
354                let types_vec = vec(
355                    proptest_arbitrary_interop::arb::<reth_ethereum_primitives::TxType>()
356                        .prop_map(|ty| ty as u8),
357                    len..=len,
358                );
359
360                // Map the usize values to the range 0..131072(0x20000)
361                let sizes_vec = vec(proptest::num::usize::ANY.prop_map(|x| x % 131072), len..=len);
362                let hashes_vec = vec(any::<B256>(), len..=len);
363
364                (types_vec, sizes_vec, hashes_vec)
365            })
366            .prop_map(|(types, sizes, hashes)| Self { types, sizes, hashes })
367            .boxed()
368    }
369
370    type Strategy = proptest::prelude::BoxedStrategy<Self>;
371}
372
373impl NewPooledTransactionHashes68 {
374    /// Returns an iterator over tx hashes zipped with corresponding metadata.
375    pub fn metadata_iter(&self) -> impl Iterator<Item = (&B256, (u8, usize))> {
376        self.hashes.iter().zip(self.types.iter().copied().zip(self.sizes.iter().copied()))
377    }
378
379    /// Appends a transaction
380    pub fn push<T: SignedTransaction>(&mut self, tx: &T) {
381        self.hashes.push(*tx.tx_hash());
382        self.sizes.push(tx.encode_2718_len());
383        self.types.push(tx.ty());
384    }
385
386    /// Appends the provided transactions
387    pub fn extend<'a, T: SignedTransaction>(&mut self, txs: impl IntoIterator<Item = &'a T>) {
388        for tx in txs {
389            self.push(tx);
390        }
391    }
392
393    /// Consumes and appends a transaction
394    pub fn with_transaction<T: SignedTransaction>(mut self, tx: &T) -> Self {
395        self.push(tx);
396        self
397    }
398
399    /// Consumes and appends the provided transactions
400    pub fn with_transactions<'a, T: SignedTransaction>(
401        mut self,
402        txs: impl IntoIterator<Item = &'a T>,
403    ) -> Self {
404        self.extend(txs);
405        self
406    }
407}
408
409impl Encodable for NewPooledTransactionHashes68 {
410    fn encode(&self, out: &mut dyn bytes::BufMut) {
411        #[derive(RlpEncodable)]
412        struct EncodableNewPooledTransactionHashes68<'a> {
413            types: &'a [u8],
414            sizes: &'a Vec<usize>,
415            hashes: &'a Vec<B256>,
416        }
417
418        let encodable = EncodableNewPooledTransactionHashes68 {
419            types: &self.types[..],
420            sizes: &self.sizes,
421            hashes: &self.hashes,
422        };
423
424        encodable.encode(out);
425    }
426    fn length(&self) -> usize {
427        #[derive(RlpEncodable)]
428        struct EncodableNewPooledTransactionHashes68<'a> {
429            types: &'a [u8],
430            sizes: &'a Vec<usize>,
431            hashes: &'a Vec<B256>,
432        }
433
434        let encodable = EncodableNewPooledTransactionHashes68 {
435            types: &self.types[..],
436            sizes: &self.sizes,
437            hashes: &self.hashes,
438        };
439
440        encodable.length()
441    }
442}
443
444impl Decodable for NewPooledTransactionHashes68 {
445    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
446        #[derive(RlpDecodable)]
447        struct EncodableNewPooledTransactionHashes68 {
448            types: Bytes,
449            sizes: Vec<usize>,
450            hashes: Vec<B256>,
451        }
452
453        let encodable = EncodableNewPooledTransactionHashes68::decode(buf)?;
454        let msg = Self {
455            types: encodable.types.into(),
456            sizes: encodable.sizes,
457            hashes: encodable.hashes,
458        };
459
460        if msg.hashes.len() != msg.types.len() {
461            return Err(alloy_rlp::Error::ListLengthMismatch {
462                expected: msg.hashes.len(),
463                got: msg.types.len(),
464            })
465        }
466        if msg.hashes.len() != msg.sizes.len() {
467            return Err(alloy_rlp::Error::ListLengthMismatch {
468                expected: msg.hashes.len(),
469                got: msg.sizes.len(),
470            })
471        }
472
473        Ok(msg)
474    }
475}
476
477/// Validation pass that checks for unique transaction hashes.
478pub trait DedupPayload {
479    /// Value type in [`PartiallyValidData`] map.
480    type Value;
481
482    /// The payload contains no entries.
483    fn is_empty(&self) -> bool;
484
485    /// Returns the number of entries.
486    fn len(&self) -> usize;
487
488    /// Consumes self, returning an iterator over hashes in payload.
489    fn dedup(self) -> PartiallyValidData<Self::Value>;
490}
491
492/// Value in [`PartiallyValidData`] map obtained from an announcement.
493pub type Eth68TxMetadata = Option<(u8, usize)>;
494
495impl DedupPayload for NewPooledTransactionHashes {
496    type Value = Eth68TxMetadata;
497
498    fn is_empty(&self) -> bool {
499        self.is_empty()
500    }
501
502    fn len(&self) -> usize {
503        self.len()
504    }
505
506    fn dedup(self) -> PartiallyValidData<Self::Value> {
507        match self {
508            Self::Eth66(msg) => msg.dedup(),
509            Self::Eth68(msg) => msg.dedup(),
510        }
511    }
512}
513
514impl DedupPayload for NewPooledTransactionHashes68 {
515    type Value = Eth68TxMetadata;
516
517    fn is_empty(&self) -> bool {
518        self.hashes.is_empty()
519    }
520
521    fn len(&self) -> usize {
522        self.hashes.len()
523    }
524
525    fn dedup(self) -> PartiallyValidData<Self::Value> {
526        let Self { hashes, mut sizes, mut types } = self;
527
528        let mut deduped_data = HashMap::with_capacity_and_hasher(hashes.len(), Default::default());
529
530        for hash in hashes.into_iter().rev() {
531            if let (Some(ty), Some(size)) = (types.pop(), sizes.pop()) {
532                deduped_data.insert(hash, Some((ty, size)));
533            }
534        }
535
536        PartiallyValidData::from_raw_data_eth68(deduped_data)
537    }
538}
539
540impl DedupPayload for NewPooledTransactionHashes66 {
541    type Value = Eth68TxMetadata;
542
543    fn is_empty(&self) -> bool {
544        self.0.is_empty()
545    }
546
547    fn len(&self) -> usize {
548        self.0.len()
549    }
550
551    fn dedup(self) -> PartiallyValidData<Self::Value> {
552        let Self(hashes) = self;
553
554        let mut deduped_data = HashMap::with_capacity_and_hasher(hashes.len(), Default::default());
555
556        let noop_value: Eth68TxMetadata = None;
557
558        for hash in hashes.into_iter().rev() {
559            deduped_data.insert(hash, noop_value);
560        }
561
562        PartiallyValidData::from_raw_data_eth66(deduped_data)
563    }
564}
565
566/// Interface for handling mempool message data. Used in various filters in pipelines in
567/// `TransactionsManager` and in queries to `TransactionPool`.
568pub trait HandleMempoolData {
569    /// The announcement contains no entries.
570    fn is_empty(&self) -> bool;
571
572    /// Returns the number of entries.
573    fn len(&self) -> usize;
574
575    /// Retain only entries for which the hash in the entry satisfies a given predicate.
576    fn retain_by_hash(&mut self, f: impl FnMut(&TxHash) -> bool);
577}
578
579/// Extension of [`HandleMempoolData`] interface, for mempool messages that are versioned.
580pub trait HandleVersionedMempoolData {
581    /// Returns the announcement version, either [`Eth66`](EthVersion::Eth66) or
582    /// [`Eth68`](EthVersion::Eth68).
583    fn msg_version(&self) -> EthVersion;
584}
585
586impl<T: SignedTransaction> HandleMempoolData for Vec<T> {
587    fn is_empty(&self) -> bool {
588        self.is_empty()
589    }
590
591    fn len(&self) -> usize {
592        self.len()
593    }
594
595    fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
596        self.retain(|tx| f(tx.tx_hash()))
597    }
598}
599
600macro_rules! handle_mempool_data_map_impl {
601    ($data_ty:ty, $(<$generic:ident>)?) => {
602        impl$(<$generic>)? HandleMempoolData for $data_ty {
603            fn is_empty(&self) -> bool {
604                self.data.is_empty()
605            }
606
607            fn len(&self) -> usize {
608                self.data.len()
609            }
610
611            fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
612                self.data.retain(|hash, _| f(hash));
613            }
614        }
615    };
616}
617
618/// Data that has passed an initial validation pass that is not specific to any mempool message
619/// type.
620#[derive(Debug, Deref, DerefMut, IntoIterator)]
621pub struct PartiallyValidData<V> {
622    #[deref]
623    #[deref_mut]
624    #[into_iterator]
625    data: HashMap<TxHash, V>,
626    version: Option<EthVersion>,
627}
628
629handle_mempool_data_map_impl!(PartiallyValidData<V>, <V>);
630
631impl<V> PartiallyValidData<V> {
632    /// Wraps raw data.
633    pub const fn from_raw_data(data: HashMap<TxHash, V>, version: Option<EthVersion>) -> Self {
634        Self { data, version }
635    }
636
637    /// Wraps raw data with version [`EthVersion::Eth68`].
638    pub const fn from_raw_data_eth68(data: HashMap<TxHash, V>) -> Self {
639        Self::from_raw_data(data, Some(EthVersion::Eth68))
640    }
641
642    /// Wraps raw data with version [`EthVersion::Eth66`].
643    pub const fn from_raw_data_eth66(data: HashMap<TxHash, V>) -> Self {
644        Self::from_raw_data(data, Some(EthVersion::Eth66))
645    }
646
647    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth68`](EthVersion::Eth68)
648    /// announcement.
649    pub fn empty_eth68() -> Self {
650        Self::from_raw_data_eth68(HashMap::default())
651    }
652
653    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth66`](EthVersion::Eth66)
654    /// announcement.
655    pub fn empty_eth66() -> Self {
656        Self::from_raw_data_eth66(HashMap::default())
657    }
658
659    /// Returns the version of the message this data was received in if different versions of the
660    /// message exists, either [`Eth66`](EthVersion::Eth66) or [`Eth68`](EthVersion::Eth68).
661    pub const fn msg_version(&self) -> Option<EthVersion> {
662        self.version
663    }
664
665    /// Destructs returning the validated data.
666    pub fn into_data(self) -> HashMap<TxHash, V> {
667        self.data
668    }
669}
670
671/// Partially validated data from an announcement or a
672/// [`PooledTransactions`](crate::PooledTransactions) response.
673#[derive(Debug, Deref, DerefMut, IntoIterator, From)]
674pub struct ValidAnnouncementData {
675    #[deref]
676    #[deref_mut]
677    #[into_iterator]
678    data: HashMap<TxHash, Eth68TxMetadata>,
679    version: EthVersion,
680}
681
682handle_mempool_data_map_impl!(ValidAnnouncementData,);
683
684impl ValidAnnouncementData {
685    /// Destructs returning only the valid hashes and the announcement message version. Caution! If
686    /// this is [`Eth68`](EthVersion::Eth68) announcement data, this drops the metadata.
687    pub fn into_request_hashes(self) -> (RequestTxHashes, EthVersion) {
688        let hashes = self.data.into_keys().collect::<HashSet<_>>();
689
690        (RequestTxHashes::new(hashes), self.version)
691    }
692
693    /// Conversion from [`PartiallyValidData`] from an announcement. Note! [`PartiallyValidData`]
694    /// from an announcement, should have some [`EthVersion`]. Panics if [`PartiallyValidData`] has
695    /// version set to `None`.
696    pub fn from_partially_valid_data(data: PartiallyValidData<Eth68TxMetadata>) -> Self {
697        let PartiallyValidData { data, version } = data;
698
699        let version = version.expect("should have eth version for conversion");
700
701        Self { data, version }
702    }
703
704    /// Destructs returning the validated data.
705    pub fn into_data(self) -> HashMap<TxHash, Eth68TxMetadata> {
706        self.data
707    }
708}
709
710impl HandleVersionedMempoolData for ValidAnnouncementData {
711    fn msg_version(&self) -> EthVersion {
712        self.version
713    }
714}
715
716/// Hashes to request from a peer.
717#[derive(Debug, Default, Deref, DerefMut, IntoIterator, Constructor)]
718pub struct RequestTxHashes {
719    #[deref]
720    #[deref_mut]
721    #[into_iterator(owned, ref)]
722    hashes: HashSet<TxHash>,
723}
724
725impl RequestTxHashes {
726    /// Returns a new [`RequestTxHashes`] with given capacity for hashes. Caution! Make sure to
727    /// call [`HashSet::shrink_to_fit`] on [`RequestTxHashes`] when full, especially where it will
728    /// be stored in its entirety like in the future waiting for a
729    /// [`GetPooledTransactions`](crate::GetPooledTransactions) request to resolve.
730    pub fn with_capacity(capacity: usize) -> Self {
731        Self::new(HashSet::with_capacity_and_hasher(capacity, Default::default()))
732    }
733
734    /// Returns an new empty instance.
735    fn empty() -> Self {
736        Self::new(HashSet::default())
737    }
738
739    /// Retains the given number of elements, returning and iterator over the rest.
740    pub fn retain_count(&mut self, count: usize) -> Self {
741        let rest_capacity = self.hashes.len().saturating_sub(count);
742        if rest_capacity == 0 {
743            return Self::empty()
744        }
745        let mut rest = Self::with_capacity(rest_capacity);
746
747        let mut i = 0;
748        self.hashes.retain(|hash| {
749            if i >= count {
750                rest.insert(*hash);
751                return false
752            }
753            i += 1;
754
755            true
756        });
757
758        rest
759    }
760}
761
762impl FromIterator<(TxHash, Eth68TxMetadata)> for RequestTxHashes {
763    fn from_iter<I: IntoIterator<Item = (TxHash, Eth68TxMetadata)>>(iter: I) -> Self {
764        Self::new(iter.into_iter().map(|(hash, _)| hash).collect())
765    }
766}
767
768/// The earliest block, the latest block and hash of the latest block which can be provided.
769/// See [BlockRangeUpdate](https://github.com/ethereum/devp2p/blob/master/caps/eth.md#blockrangeupdate-0x11).
770#[derive(Clone, Debug, PartialEq, Eq, Default, RlpEncodable, RlpDecodable)]
771#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
772#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))]
773pub struct BlockRangeUpdate {
774    /// The earliest block which is available.
775    pub earliest: u64,
776    /// The latest block which is available.
777    pub latest: u64,
778    /// Latest available block's hash.
779    pub latest_hash: B256,
780}
781
782#[cfg(test)]
783mod tests {
784    use super::*;
785    use alloy_consensus::Typed2718;
786    use alloy_eips::eip2718::Encodable2718;
787    use alloy_primitives::{b256, hex, Signature, U256};
788    use reth_ethereum_primitives::{Transaction, TransactionSigned};
789    use std::str::FromStr;
790
791    /// Takes as input a struct / encoded hex message pair, ensuring that we encode to the exact hex
792    /// message, and decode to the exact struct.
793    fn test_encoding_vector<T: Encodable + Decodable + PartialEq + core::fmt::Debug>(
794        input: (T, &[u8]),
795    ) {
796        let (expected_decoded, expected_encoded) = input;
797        let mut encoded = Vec::new();
798        expected_decoded.encode(&mut encoded);
799
800        assert_eq!(hex::encode(&encoded), hex::encode(expected_encoded));
801
802        let decoded = T::decode(&mut encoded.as_ref()).unwrap();
803        assert_eq!(expected_decoded, decoded);
804    }
805
806    #[test]
807    fn can_return_latest_block() {
808        let mut blocks = NewBlockHashes(vec![BlockHashNumber { hash: B256::random(), number: 0 }]);
809        let latest = blocks.latest().unwrap();
810        assert_eq!(latest.number, 0);
811
812        blocks.0.push(BlockHashNumber { hash: B256::random(), number: 100 });
813        blocks.0.push(BlockHashNumber { hash: B256::random(), number: 2 });
814        let latest = blocks.latest().unwrap();
815        assert_eq!(latest.number, 100);
816    }
817
818    #[test]
819    fn eth_68_tx_hash_roundtrip() {
820        let vectors = vec![
821            (
822                NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] },
823                &hex!("c380c0c0")[..],
824            ),
825            (
826                NewPooledTransactionHashes68 {
827                    types: vec![0x00],
828                    sizes: vec![0x00],
829                    hashes: vec![
830                        B256::from_str(
831                            "0x0000000000000000000000000000000000000000000000000000000000000000",
832                        )
833                        .unwrap(),
834                    ],
835                },
836                &hex!(
837                    "e500c180e1a00000000000000000000000000000000000000000000000000000000000000000"
838                )[..],
839            ),
840            (
841                NewPooledTransactionHashes68 {
842                    types: vec![0x00, 0x00],
843                    sizes: vec![0x00, 0x00],
844                    hashes: vec![
845                        B256::from_str(
846                            "0x0000000000000000000000000000000000000000000000000000000000000000",
847                        )
848                        .unwrap(),
849                        B256::from_str(
850                            "0x0000000000000000000000000000000000000000000000000000000000000000",
851                        )
852                        .unwrap(),
853                    ],
854                },
855                &hex!(
856                    "f84a820000c28080f842a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000"
857                )[..],
858            ),
859            (
860                NewPooledTransactionHashes68 {
861                    types: vec![0x02],
862                    sizes: vec![0xb6],
863                    hashes: vec![
864                        B256::from_str(
865                            "0xfecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124",
866                        )
867                        .unwrap(),
868                    ],
869                },
870                &hex!(
871                    "e602c281b6e1a0fecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124"
872                )[..],
873            ),
874            (
875                NewPooledTransactionHashes68 {
876                    types: vec![0xff, 0xff],
877                    sizes: vec![0xffffffff, 0xffffffff],
878                    hashes: vec![
879                        B256::from_str(
880                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
881                        )
882                        .unwrap(),
883                        B256::from_str(
884                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
885                        )
886                        .unwrap(),
887                    ],
888                },
889                &hex!(
890                    "f85282ffffca84ffffffff84fffffffff842a0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffa0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
891                )[..],
892            ),
893            (
894                NewPooledTransactionHashes68 {
895                    types: vec![0xff, 0xff],
896                    sizes: vec![0xffffffff, 0xffffffff],
897                    hashes: vec![
898                        B256::from_str(
899                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
900                        )
901                        .unwrap(),
902                        B256::from_str(
903                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
904                        )
905                        .unwrap(),
906                    ],
907                },
908                &hex!(
909                    "f85282ffffca84ffffffff84fffffffff842a0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafea0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe"
910                )[..],
911            ),
912            (
913                NewPooledTransactionHashes68 {
914                    types: vec![0x10, 0x10],
915                    sizes: vec![0xdeadc0de, 0xdeadc0de],
916                    hashes: vec![
917                        B256::from_str(
918                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
919                        )
920                        .unwrap(),
921                        B256::from_str(
922                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
923                        )
924                        .unwrap(),
925                    ],
926                },
927                &hex!(
928                    "f852821010ca84deadc0de84deadc0def842a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2"
929                )[..],
930            ),
931            (
932                NewPooledTransactionHashes68 {
933                    types: vec![0x6f, 0x6f],
934                    sizes: vec![0x7fffffff, 0x7fffffff],
935                    hashes: vec![
936                        B256::from_str(
937                            "0x0000000000000000000000000000000000000000000000000000000000000002",
938                        )
939                        .unwrap(),
940                        B256::from_str(
941                            "0x0000000000000000000000000000000000000000000000000000000000000002",
942                        )
943                        .unwrap(),
944                    ],
945                },
946                &hex!(
947                    "f852826f6fca847fffffff847ffffffff842a00000000000000000000000000000000000000000000000000000000000000002a00000000000000000000000000000000000000000000000000000000000000002"
948                )[..],
949            ),
950        ];
951
952        for vector in vectors {
953            test_encoding_vector(vector);
954        }
955    }
956
957    #[test]
958    fn request_hashes_retain_count_keep_subset() {
959        let mut hashes = RequestTxHashes::new(
960            [
961                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
962                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
963                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
964                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
965                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
966            ]
967            .into_iter()
968            .collect::<HashSet<_>>(),
969        );
970
971        let rest = hashes.retain_count(3);
972
973        assert_eq!(3, hashes.len());
974        assert_eq!(2, rest.len());
975    }
976
977    #[test]
978    fn request_hashes_retain_count_keep_all() {
979        let mut hashes = RequestTxHashes::new(
980            [
981                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
982                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
983                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
984                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
985                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
986            ]
987            .into_iter()
988            .collect::<HashSet<_>>(),
989        );
990
991        let _ = hashes.retain_count(6);
992
993        assert_eq!(5, hashes.len());
994    }
995
996    #[test]
997    fn split_request_hashes_keep_none() {
998        let mut hashes = RequestTxHashes::new(
999            [
1000                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1001                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1002                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1003                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1004                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1005            ]
1006            .into_iter()
1007            .collect::<HashSet<_>>(),
1008        );
1009
1010        let rest = hashes.retain_count(0);
1011
1012        assert_eq!(0, hashes.len());
1013        assert_eq!(5, rest.len());
1014    }
1015
1016    fn signed_transaction() -> impl SignedTransaction {
1017        TransactionSigned::new_unhashed(
1018            Transaction::Legacy(Default::default()),
1019            Signature::new(
1020                U256::from_str(
1021                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c12",
1022                )
1023                .unwrap(),
1024                U256::from_str(
1025                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c10",
1026                )
1027                .unwrap(),
1028                false,
1029            ),
1030        )
1031    }
1032
1033    #[test]
1034    fn test_pooled_tx_hashes_68_push() {
1035        let tx = signed_transaction();
1036        let mut tx_hashes =
1037            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1038        tx_hashes.push(&tx);
1039        assert_eq!(tx_hashes.types.len(), 1);
1040        assert_eq!(tx_hashes.sizes.len(), 1);
1041        assert_eq!(tx_hashes.hashes.len(), 1);
1042        assert_eq!(tx_hashes.types[0], tx.ty());
1043        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1044        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1045    }
1046
1047    #[test]
1048    fn test_pooled_tx_hashes_68_extend() {
1049        let tx = signed_transaction();
1050        let txs = vec![tx.clone(), tx.clone()];
1051        let mut tx_hashes =
1052            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1053        tx_hashes.extend(&txs);
1054        assert_eq!(tx_hashes.types.len(), 2);
1055        assert_eq!(tx_hashes.sizes.len(), 2);
1056        assert_eq!(tx_hashes.hashes.len(), 2);
1057        assert_eq!(tx_hashes.types[0], tx.ty());
1058        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1059        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1060        assert_eq!(tx_hashes.types[1], tx.ty());
1061        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1062        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1063    }
1064
1065    #[test]
1066    fn test_pooled_tx_hashes_68_with_transaction() {
1067        let tx = signed_transaction();
1068        let tx_hashes =
1069            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1070                .with_transaction(&tx);
1071        assert_eq!(tx_hashes.types.len(), 1);
1072        assert_eq!(tx_hashes.sizes.len(), 1);
1073        assert_eq!(tx_hashes.hashes.len(), 1);
1074        assert_eq!(tx_hashes.types[0], tx.ty());
1075        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1076        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1077    }
1078
1079    #[test]
1080    fn test_pooled_tx_hashes_68_with_transactions() {
1081        let tx = signed_transaction();
1082        let txs = vec![tx.clone(), tx.clone()];
1083        let tx_hashes =
1084            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1085                .with_transactions(&txs);
1086        assert_eq!(tx_hashes.types.len(), 2);
1087        assert_eq!(tx_hashes.sizes.len(), 2);
1088        assert_eq!(tx_hashes.hashes.len(), 2);
1089        assert_eq!(tx_hashes.types[0], tx.ty());
1090        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1091        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1092        assert_eq!(tx_hashes.types[1], tx.ty());
1093        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1094        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1095    }
1096}
reth_eth_wire_types/broadcast.rs

reth_eth_wire_types/
broadcast.rs
