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 | EthVersion::Eth70)
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    /// Shrinks the capacity of the message vectors as much as possible.
413    pub fn shrink_to_fit(&mut self) {
414        self.hashes.shrink_to_fit();
415        self.sizes.shrink_to_fit();
416        self.types.shrink_to_fit()
417    }
418
419    /// Consumes and appends a transaction
420    pub fn with_transaction<T: SignedTransaction>(mut self, tx: &T) -> Self {
421        self.push(tx);
422        self
423    }
424
425    /// Consumes and appends the provided transactions
426    pub fn with_transactions<'a, T: SignedTransaction>(
427        mut self,
428        txs: impl IntoIterator<Item = &'a T>,
429    ) -> Self {
430        self.extend(txs);
431        self
432    }
433}
434
435impl Encodable for NewPooledTransactionHashes68 {
436    fn encode(&self, out: &mut dyn bytes::BufMut) {
437        #[derive(RlpEncodable)]
438        struct EncodableNewPooledTransactionHashes68<'a> {
439            types: &'a [u8],
440            sizes: &'a Vec<usize>,
441            hashes: &'a Vec<B256>,
442        }
443
444        let encodable = EncodableNewPooledTransactionHashes68 {
445            types: &self.types[..],
446            sizes: &self.sizes,
447            hashes: &self.hashes,
448        };
449
450        encodable.encode(out);
451    }
452    fn length(&self) -> usize {
453        #[derive(RlpEncodable)]
454        struct EncodableNewPooledTransactionHashes68<'a> {
455            types: &'a [u8],
456            sizes: &'a Vec<usize>,
457            hashes: &'a Vec<B256>,
458        }
459
460        let encodable = EncodableNewPooledTransactionHashes68 {
461            types: &self.types[..],
462            sizes: &self.sizes,
463            hashes: &self.hashes,
464        };
465
466        encodable.length()
467    }
468}
469
470impl Decodable for NewPooledTransactionHashes68 {
471    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
472        #[derive(RlpDecodable)]
473        struct EncodableNewPooledTransactionHashes68 {
474            types: Bytes,
475            sizes: Vec<usize>,
476            hashes: Vec<B256>,
477        }
478
479        let encodable = EncodableNewPooledTransactionHashes68::decode(buf)?;
480        let msg = Self {
481            types: encodable.types.into(),
482            sizes: encodable.sizes,
483            hashes: encodable.hashes,
484        };
485
486        if msg.hashes.len() != msg.types.len() {
487            return Err(alloy_rlp::Error::ListLengthMismatch {
488                expected: msg.hashes.len(),
489                got: msg.types.len(),
490            })
491        }
492        if msg.hashes.len() != msg.sizes.len() {
493            return Err(alloy_rlp::Error::ListLengthMismatch {
494                expected: msg.hashes.len(),
495                got: msg.sizes.len(),
496            })
497        }
498
499        Ok(msg)
500    }
501}
502
503/// Validation pass that checks for unique transaction hashes.
504pub trait DedupPayload {
505    /// Value type in [`PartiallyValidData`] map.
506    type Value;
507
508    /// The payload contains no entries.
509    fn is_empty(&self) -> bool;
510
511    /// Returns the number of entries.
512    fn len(&self) -> usize;
513
514    /// Consumes self, returning an iterator over hashes in payload.
515    fn dedup(self) -> PartiallyValidData<Self::Value>;
516}
517
518/// Value in [`PartiallyValidData`] map obtained from an announcement.
519pub type Eth68TxMetadata = Option<(u8, usize)>;
520
521impl DedupPayload for NewPooledTransactionHashes {
522    type Value = Eth68TxMetadata;
523
524    fn is_empty(&self) -> bool {
525        self.is_empty()
526    }
527
528    fn len(&self) -> usize {
529        self.len()
530    }
531
532    fn dedup(self) -> PartiallyValidData<Self::Value> {
533        match self {
534            Self::Eth66(msg) => msg.dedup(),
535            Self::Eth68(msg) => msg.dedup(),
536        }
537    }
538}
539
540impl DedupPayload for NewPooledTransactionHashes68 {
541    type Value = Eth68TxMetadata;
542
543    fn is_empty(&self) -> bool {
544        self.hashes.is_empty()
545    }
546
547    fn len(&self) -> usize {
548        self.hashes.len()
549    }
550
551    fn dedup(self) -> PartiallyValidData<Self::Value> {
552        let Self { hashes, mut sizes, mut types } = self;
553
554        let mut deduped_data = HashMap::with_capacity_and_hasher(hashes.len(), Default::default());
555
556        for hash in hashes.into_iter().rev() {
557            if let (Some(ty), Some(size)) = (types.pop(), sizes.pop()) {
558                deduped_data.insert(hash, Some((ty, size)));
559            }
560        }
561
562        PartiallyValidData::from_raw_data_eth68(deduped_data)
563    }
564}
565
566impl DedupPayload for NewPooledTransactionHashes66 {
567    type Value = Eth68TxMetadata;
568
569    fn is_empty(&self) -> bool {
570        self.0.is_empty()
571    }
572
573    fn len(&self) -> usize {
574        self.0.len()
575    }
576
577    fn dedup(self) -> PartiallyValidData<Self::Value> {
578        let Self(hashes) = self;
579
580        let mut deduped_data = HashMap::with_capacity_and_hasher(hashes.len(), Default::default());
581
582        let noop_value: Eth68TxMetadata = None;
583
584        for hash in hashes.into_iter().rev() {
585            deduped_data.insert(hash, noop_value);
586        }
587
588        PartiallyValidData::from_raw_data_eth66(deduped_data)
589    }
590}
591
592/// Interface for handling mempool message data. Used in various filters in pipelines in
593/// `TransactionsManager` and in queries to `TransactionPool`.
594pub trait HandleMempoolData {
595    /// The announcement contains no entries.
596    fn is_empty(&self) -> bool;
597
598    /// Returns the number of entries.
599    fn len(&self) -> usize;
600
601    /// Retain only entries for which the hash in the entry satisfies a given predicate.
602    fn retain_by_hash(&mut self, f: impl FnMut(&TxHash) -> bool);
603}
604
605/// Extension of [`HandleMempoolData`] interface, for mempool messages that are versioned.
606pub trait HandleVersionedMempoolData {
607    /// Returns the announcement version, either [`Eth66`](EthVersion::Eth66) or
608    /// [`Eth68`](EthVersion::Eth68).
609    fn msg_version(&self) -> EthVersion;
610}
611
612impl<T: SignedTransaction> HandleMempoolData for Vec<T> {
613    fn is_empty(&self) -> bool {
614        self.is_empty()
615    }
616
617    fn len(&self) -> usize {
618        self.len()
619    }
620
621    fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
622        self.retain(|tx| f(tx.tx_hash()))
623    }
624}
625
626macro_rules! handle_mempool_data_map_impl {
627    ($data_ty:ty, $(<$generic:ident>)?) => {
628        impl$(<$generic>)? HandleMempoolData for $data_ty {
629            fn is_empty(&self) -> bool {
630                self.data.is_empty()
631            }
632
633            fn len(&self) -> usize {
634                self.data.len()
635            }
636
637            fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
638                self.data.retain(|hash, _| f(hash));
639            }
640        }
641    };
642}
643
644/// Data that has passed an initial validation pass that is not specific to any mempool message
645/// type.
646#[derive(Debug, Deref, DerefMut, IntoIterator)]
647pub struct PartiallyValidData<V> {
648    #[deref]
649    #[deref_mut]
650    #[into_iterator]
651    data: HashMap<TxHash, V>,
652    version: Option<EthVersion>,
653}
654
655handle_mempool_data_map_impl!(PartiallyValidData<V>, <V>);
656
657impl<V> PartiallyValidData<V> {
658    /// Wraps raw data.
659    pub const fn from_raw_data(data: HashMap<TxHash, V>, version: Option<EthVersion>) -> Self {
660        Self { data, version }
661    }
662
663    /// Wraps raw data with version [`EthVersion::Eth68`].
664    pub const fn from_raw_data_eth68(data: HashMap<TxHash, V>) -> Self {
665        Self::from_raw_data(data, Some(EthVersion::Eth68))
666    }
667
668    /// Wraps raw data with version [`EthVersion::Eth66`].
669    pub const fn from_raw_data_eth66(data: HashMap<TxHash, V>) -> Self {
670        Self::from_raw_data(data, Some(EthVersion::Eth66))
671    }
672
673    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth68`](EthVersion::Eth68)
674    /// announcement.
675    pub fn empty_eth68() -> Self {
676        Self::from_raw_data_eth68(HashMap::default())
677    }
678
679    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth66`](EthVersion::Eth66)
680    /// announcement.
681    pub fn empty_eth66() -> Self {
682        Self::from_raw_data_eth66(HashMap::default())
683    }
684
685    /// Returns the version of the message this data was received in if different versions of the
686    /// message exists, either [`Eth66`](EthVersion::Eth66) or [`Eth68`](EthVersion::Eth68).
687    pub const fn msg_version(&self) -> Option<EthVersion> {
688        self.version
689    }
690
691    /// Destructs returning the validated data.
692    pub fn into_data(self) -> HashMap<TxHash, V> {
693        self.data
694    }
695}
696
697/// Partially validated data from an announcement or a
698/// [`PooledTransactions`](crate::PooledTransactions) response.
699#[derive(Debug, Deref, DerefMut, IntoIterator, From)]
700pub struct ValidAnnouncementData {
701    #[deref]
702    #[deref_mut]
703    #[into_iterator]
704    data: HashMap<TxHash, Eth68TxMetadata>,
705    version: EthVersion,
706}
707
708handle_mempool_data_map_impl!(ValidAnnouncementData,);
709
710impl ValidAnnouncementData {
711    /// Destructs returning only the valid hashes and the announcement message version. Caution! If
712    /// this is [`Eth68`](EthVersion::Eth68) announcement data, this drops the metadata.
713    pub fn into_request_hashes(self) -> (RequestTxHashes, EthVersion) {
714        let hashes = self.data.into_keys().collect::<HashSet<_>>();
715
716        (RequestTxHashes::new(hashes), self.version)
717    }
718
719    /// Conversion from [`PartiallyValidData`] from an announcement. Note! [`PartiallyValidData`]
720    /// from an announcement, should have some [`EthVersion`]. Panics if [`PartiallyValidData`] has
721    /// version set to `None`.
722    pub fn from_partially_valid_data(data: PartiallyValidData<Eth68TxMetadata>) -> Self {
723        let PartiallyValidData { data, version } = data;
724
725        let version = version.expect("should have eth version for conversion");
726
727        Self { data, version }
728    }
729
730    /// Destructs returning the validated data.
731    pub fn into_data(self) -> HashMap<TxHash, Eth68TxMetadata> {
732        self.data
733    }
734}
735
736impl HandleVersionedMempoolData for ValidAnnouncementData {
737    fn msg_version(&self) -> EthVersion {
738        self.version
739    }
740}
741
742/// Hashes to request from a peer.
743#[derive(Debug, Default, Deref, DerefMut, IntoIterator, Constructor)]
744pub struct RequestTxHashes {
745    #[deref]
746    #[deref_mut]
747    #[into_iterator(owned, ref)]
748    hashes: HashSet<TxHash>,
749}
750
751impl RequestTxHashes {
752    /// Returns a new [`RequestTxHashes`] with given capacity for hashes. Caution! Make sure to
753    /// call [`HashSet::shrink_to_fit`] on [`RequestTxHashes`] when full, especially where it will
754    /// be stored in its entirety like in the future waiting for a
755    /// [`GetPooledTransactions`](crate::GetPooledTransactions) request to resolve.
756    pub fn with_capacity(capacity: usize) -> Self {
757        Self::new(HashSet::with_capacity_and_hasher(capacity, Default::default()))
758    }
759
760    /// Returns an new empty instance.
761    fn empty() -> Self {
762        Self::new(HashSet::default())
763    }
764
765    /// Retains the given number of elements, returning and iterator over the rest.
766    pub fn retain_count(&mut self, count: usize) -> Self {
767        let rest_capacity = self.hashes.len().saturating_sub(count);
768        if rest_capacity == 0 {
769            return Self::empty()
770        }
771        let mut rest = Self::with_capacity(rest_capacity);
772
773        let mut i = 0;
774        self.hashes.retain(|hash| {
775            if i >= count {
776                rest.insert(*hash);
777                return false
778            }
779            i += 1;
780
781            true
782        });
783
784        rest
785    }
786}
787
788impl FromIterator<(TxHash, Eth68TxMetadata)> for RequestTxHashes {
789    fn from_iter<I: IntoIterator<Item = (TxHash, Eth68TxMetadata)>>(iter: I) -> Self {
790        Self::new(iter.into_iter().map(|(hash, _)| hash).collect())
791    }
792}
793
794/// The earliest block, the latest block and hash of the latest block which can be provided.
795/// See [BlockRangeUpdate](https://github.com/ethereum/devp2p/blob/master/caps/eth.md#blockrangeupdate-0x11).
796#[derive(Clone, Debug, PartialEq, Eq, Default, RlpEncodable, RlpDecodable)]
797#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
798#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))]
799pub struct BlockRangeUpdate {
800    /// The earliest block which is available.
801    pub earliest: u64,
802    /// The latest block which is available.
803    pub latest: u64,
804    /// Latest available block's hash.
805    pub latest_hash: B256,
806}
807
808#[cfg(test)]
809mod tests {
810    use super::*;
811    use alloy_consensus::{transaction::TxHashRef, Typed2718};
812    use alloy_eips::eip2718::Encodable2718;
813    use alloy_primitives::{b256, hex, Signature, U256};
814    use reth_ethereum_primitives::{Transaction, TransactionSigned};
815    use std::str::FromStr;
816
817    /// Takes as input a struct / encoded hex message pair, ensuring that we encode to the exact hex
818    /// message, and decode to the exact struct.
819    fn test_encoding_vector<T: Encodable + Decodable + PartialEq + core::fmt::Debug>(
820        input: (T, &[u8]),
821    ) {
822        let (expected_decoded, expected_encoded) = input;
823        let mut encoded = Vec::new();
824        expected_decoded.encode(&mut encoded);
825
826        assert_eq!(hex::encode(&encoded), hex::encode(expected_encoded));
827
828        let decoded = T::decode(&mut encoded.as_ref()).unwrap();
829        assert_eq!(expected_decoded, decoded);
830    }
831
832    #[test]
833    fn can_return_latest_block() {
834        let mut blocks = NewBlockHashes(vec![BlockHashNumber { hash: B256::random(), number: 0 }]);
835        let latest = blocks.latest().unwrap();
836        assert_eq!(latest.number, 0);
837
838        blocks.0.push(BlockHashNumber { hash: B256::random(), number: 100 });
839        blocks.0.push(BlockHashNumber { hash: B256::random(), number: 2 });
840        let latest = blocks.latest().unwrap();
841        assert_eq!(latest.number, 100);
842    }
843
844    #[test]
845    fn eth_68_tx_hash_roundtrip() {
846        let vectors = vec![
847            (
848                NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] },
849                &hex!("c380c0c0")[..],
850            ),
851            (
852                NewPooledTransactionHashes68 {
853                    types: vec![0x00],
854                    sizes: vec![0x00],
855                    hashes: vec![
856                        B256::from_str(
857                            "0x0000000000000000000000000000000000000000000000000000000000000000",
858                        )
859                        .unwrap(),
860                    ],
861                },
862                &hex!(
863                    "e500c180e1a00000000000000000000000000000000000000000000000000000000000000000"
864                )[..],
865            ),
866            (
867                NewPooledTransactionHashes68 {
868                    types: vec![0x00, 0x00],
869                    sizes: vec![0x00, 0x00],
870                    hashes: vec![
871                        B256::from_str(
872                            "0x0000000000000000000000000000000000000000000000000000000000000000",
873                        )
874                        .unwrap(),
875                        B256::from_str(
876                            "0x0000000000000000000000000000000000000000000000000000000000000000",
877                        )
878                        .unwrap(),
879                    ],
880                },
881                &hex!(
882                    "f84a820000c28080f842a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000"
883                )[..],
884            ),
885            (
886                NewPooledTransactionHashes68 {
887                    types: vec![0x02],
888                    sizes: vec![0xb6],
889                    hashes: vec![
890                        B256::from_str(
891                            "0xfecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124",
892                        )
893                        .unwrap(),
894                    ],
895                },
896                &hex!(
897                    "e602c281b6e1a0fecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124"
898                )[..],
899            ),
900            (
901                NewPooledTransactionHashes68 {
902                    types: vec![0xff, 0xff],
903                    sizes: vec![0xffffffff, 0xffffffff],
904                    hashes: vec![
905                        B256::from_str(
906                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
907                        )
908                        .unwrap(),
909                        B256::from_str(
910                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
911                        )
912                        .unwrap(),
913                    ],
914                },
915                &hex!(
916                    "f85282ffffca84ffffffff84fffffffff842a0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffa0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
917                )[..],
918            ),
919            (
920                NewPooledTransactionHashes68 {
921                    types: vec![0xff, 0xff],
922                    sizes: vec![0xffffffff, 0xffffffff],
923                    hashes: vec![
924                        B256::from_str(
925                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
926                        )
927                        .unwrap(),
928                        B256::from_str(
929                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
930                        )
931                        .unwrap(),
932                    ],
933                },
934                &hex!(
935                    "f85282ffffca84ffffffff84fffffffff842a0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafea0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe"
936                )[..],
937            ),
938            (
939                NewPooledTransactionHashes68 {
940                    types: vec![0x10, 0x10],
941                    sizes: vec![0xdeadc0de, 0xdeadc0de],
942                    hashes: vec![
943                        B256::from_str(
944                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
945                        )
946                        .unwrap(),
947                        B256::from_str(
948                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
949                        )
950                        .unwrap(),
951                    ],
952                },
953                &hex!(
954                    "f852821010ca84deadc0de84deadc0def842a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2"
955                )[..],
956            ),
957            (
958                NewPooledTransactionHashes68 {
959                    types: vec![0x6f, 0x6f],
960                    sizes: vec![0x7fffffff, 0x7fffffff],
961                    hashes: vec![
962                        B256::from_str(
963                            "0x0000000000000000000000000000000000000000000000000000000000000002",
964                        )
965                        .unwrap(),
966                        B256::from_str(
967                            "0x0000000000000000000000000000000000000000000000000000000000000002",
968                        )
969                        .unwrap(),
970                    ],
971                },
972                &hex!(
973                    "f852826f6fca847fffffff847ffffffff842a00000000000000000000000000000000000000000000000000000000000000002a00000000000000000000000000000000000000000000000000000000000000002"
974                )[..],
975            ),
976        ];
977
978        for vector in vectors {
979            test_encoding_vector(vector);
980        }
981    }
982
983    #[test]
984    fn request_hashes_retain_count_keep_subset() {
985        let mut hashes = RequestTxHashes::new(
986            [
987                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
988                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
989                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
990                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
991                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
992            ]
993            .into_iter()
994            .collect::<HashSet<_>>(),
995        );
996
997        let rest = hashes.retain_count(3);
998
999        assert_eq!(3, hashes.len());
1000        assert_eq!(2, rest.len());
1001    }
1002
1003    #[test]
1004    fn request_hashes_retain_count_keep_all() {
1005        let mut hashes = RequestTxHashes::new(
1006            [
1007                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1008                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1009                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1010                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1011                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1012            ]
1013            .into_iter()
1014            .collect::<HashSet<_>>(),
1015        );
1016
1017        let _ = hashes.retain_count(6);
1018
1019        assert_eq!(5, hashes.len());
1020    }
1021
1022    #[test]
1023    fn split_request_hashes_keep_none() {
1024        let mut hashes = RequestTxHashes::new(
1025            [
1026                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1027                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1028                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1029                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1030                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1031            ]
1032            .into_iter()
1033            .collect::<HashSet<_>>(),
1034        );
1035
1036        let rest = hashes.retain_count(0);
1037
1038        assert_eq!(0, hashes.len());
1039        assert_eq!(5, rest.len());
1040    }
1041
1042    fn signed_transaction() -> impl SignedTransaction {
1043        TransactionSigned::new_unhashed(
1044            Transaction::Legacy(Default::default()),
1045            Signature::new(
1046                U256::from_str(
1047                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c12",
1048                )
1049                .unwrap(),
1050                U256::from_str(
1051                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c10",
1052                )
1053                .unwrap(),
1054                false,
1055            ),
1056        )
1057    }
1058
1059    #[test]
1060    fn test_pooled_tx_hashes_68_push() {
1061        let tx = signed_transaction();
1062        let mut tx_hashes =
1063            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1064        tx_hashes.push(&tx);
1065        assert_eq!(tx_hashes.types.len(), 1);
1066        assert_eq!(tx_hashes.sizes.len(), 1);
1067        assert_eq!(tx_hashes.hashes.len(), 1);
1068        assert_eq!(tx_hashes.types[0], tx.ty());
1069        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1070        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1071    }
1072
1073    #[test]
1074    fn test_pooled_tx_hashes_68_extend() {
1075        let tx = signed_transaction();
1076        let txs = vec![tx.clone(), tx.clone()];
1077        let mut tx_hashes =
1078            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1079        tx_hashes.extend(&txs);
1080        assert_eq!(tx_hashes.types.len(), 2);
1081        assert_eq!(tx_hashes.sizes.len(), 2);
1082        assert_eq!(tx_hashes.hashes.len(), 2);
1083        assert_eq!(tx_hashes.types[0], tx.ty());
1084        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1085        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1086        assert_eq!(tx_hashes.types[1], tx.ty());
1087        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1088        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1089    }
1090
1091    #[test]
1092    fn test_pooled_tx_hashes_68_with_transaction() {
1093        let tx = signed_transaction();
1094        let tx_hashes =
1095            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1096                .with_transaction(&tx);
1097        assert_eq!(tx_hashes.types.len(), 1);
1098        assert_eq!(tx_hashes.sizes.len(), 1);
1099        assert_eq!(tx_hashes.hashes.len(), 1);
1100        assert_eq!(tx_hashes.types[0], tx.ty());
1101        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1102        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1103    }
1104
1105    #[test]
1106    fn test_pooled_tx_hashes_68_with_transactions() {
1107        let tx = signed_transaction();
1108        let txs = vec![tx.clone(), tx.clone()];
1109        let tx_hashes =
1110            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1111                .with_transactions(&txs);
1112        assert_eq!(tx_hashes.types.len(), 2);
1113        assert_eq!(tx_hashes.sizes.len(), 2);
1114        assert_eq!(tx_hashes.hashes.len(), 2);
1115        assert_eq!(tx_hashes.types[0], tx.ty());
1116        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1117        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1118        assert_eq!(tx_hashes.types[1], tx.ty());
1119        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1120        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1121    }
1122}
reth_eth_wire_types/broadcast.rs

reth_eth_wire_types/
broadcast.rs
