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

reth_eth_wire_types/
broadcast.rs
