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_consensus::transaction::TxHashRef;
6use alloy_eips::eip2718::Typed2718;
7use alloy_primitives::{
8    bytes::BufMut,
9    map::{B256Map, B256Set},
10    Bytes, TxHash, B128, B256, U128,
11};
12use alloy_rlp::{
13    decode_append, Decodable, Encodable, Header, RlpDecodable, RlpDecodableWrapper, RlpEncodable,
14    RlpEncodableWrapper,
15};
16use core::{fmt::Debug, mem};
17use derive_more::{Constructor, Deref, DerefMut, From, IntoIterator};
18use reth_codecs_derive::{add_arbitrary_tests, generate_tests};
19use reth_ethereum_primitives::TransactionSigned;
20use reth_primitives_traits::{sync::OnceLock, Block, InMemorySize, SignedTransaction};
21
22/// Soft limit for the number of hashes in a
23/// [`NewPooledTransactionHashes`] broadcast message.
24///
25/// Spec'd at 4096 hashes.
26///
27/// <https://github.com/ethereum/devp2p/blob/master/caps/eth.md#newpooledtransactionhashes-0x08>
28pub const SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE: usize = 4096;
29
30/// This informs peers of new blocks that have appeared on the network.
31#[derive(
32    Clone,
33    Debug,
34    PartialEq,
35    Eq,
36    RlpEncodableWrapper,
37    RlpDecodableWrapper,
38    Default,
39    Deref,
40    DerefMut,
41    IntoIterator,
42)]
43#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
44#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
45#[add_arbitrary_tests(rlp)]
46pub struct NewBlockHashes(
47    /// New block hashes and the block number for each blockhash.
48    /// Clients should request blocks using a [`GetBlockBodies`](crate::GetBlockBodies) message.
49    pub Vec<BlockHashNumber>,
50);
51
52// === impl NewBlockHashes ===
53
54impl NewBlockHashes {
55    /// Returns the latest block in the list of blocks.
56    pub fn latest(&self) -> Option<&BlockHashNumber> {
57        self.iter().max_by_key(|b| b.number)
58    }
59}
60
61/// A block hash _and_ a block number.
62#[derive(Clone, Debug, PartialEq, Eq, RlpEncodable, RlpDecodable, Default)]
63#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
64#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
65#[add_arbitrary_tests(rlp)]
66pub struct BlockHashNumber {
67    /// The block hash
68    pub hash: B256,
69    /// The block number
70    pub number: u64,
71}
72
73impl From<Vec<BlockHashNumber>> for NewBlockHashes {
74    fn from(v: Vec<BlockHashNumber>) -> Self {
75        Self(v)
76    }
77}
78
79impl From<NewBlockHashes> for Vec<BlockHashNumber> {
80    fn from(v: NewBlockHashes) -> Self {
81        v.0
82    }
83}
84
85/// A trait for block payloads transmitted through p2p.
86pub trait NewBlockPayload:
87    Encodable + Decodable + Clone + Eq + Debug + Send + Sync + Unpin + 'static
88{
89    /// The block type.
90    type Block: Block;
91
92    /// Returns a reference to the block.
93    fn block(&self) -> &Self::Block;
94}
95
96/// A new block with the current total difficulty, which includes the difficulty of the returned
97/// block.
98#[derive(Clone, Debug, PartialEq, Eq, RlpEncodable, RlpDecodable, Default)]
99#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
100#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
101pub struct NewBlock<B = reth_ethereum_primitives::Block> {
102    /// A new block.
103    pub block: B,
104    /// The current total difficulty.
105    pub td: U128,
106}
107
108impl<B: Block + 'static> NewBlockPayload for NewBlock<B> {
109    type Block = B;
110
111    fn block(&self) -> &Self::Block {
112        &self.block
113    }
114}
115
116generate_tests!(#[rlp, 25] NewBlock<reth_ethereum_primitives::Block>, EthNewBlockTests);
117
118/// This informs peers of transactions that have appeared on the network and are not yet included
119/// in a block.
120#[derive(
121    Clone,
122    Debug,
123    PartialEq,
124    Eq,
125    RlpEncodableWrapper,
126    RlpDecodableWrapper,
127    Default,
128    Deref,
129    IntoIterator,
130)]
131#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
132#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
133#[add_arbitrary_tests(rlp, 10)]
134pub struct Transactions<T = TransactionSigned>(
135    /// New transactions for the peer to include in its mempool.
136    pub Vec<T>,
137);
138
139impl<T: SignedTransaction> Transactions<T> {
140    /// Returns `true` if the list of transactions contains any blob transactions.
141    pub fn has_eip4844(&self) -> bool {
142        self.iter().any(|tx| tx.is_eip4844())
143    }
144}
145
146impl<T> From<Vec<T>> for Transactions<T> {
147    fn from(txs: Vec<T>) -> Self {
148        Self(txs)
149    }
150}
151
152impl<T> From<Transactions<T>> for Vec<T> {
153    fn from(txs: Transactions<T>) -> Self {
154        txs.0
155    }
156}
157
158impl<T: Decodable + InMemorySize> Transactions<T> {
159    /// Decodes the RLP list of transactions, stopping once the cumulative
160    /// [`InMemorySize`] of decoded transactions exceeds `memory_budget` bytes.
161    /// Any remaining transactions in the payload are skipped.
162    pub fn decode_with_memory_budget(
163        buf: &mut &[u8],
164        memory_budget: usize,
165    ) -> alloy_rlp::Result<Self> {
166        decode_list_with_memory_budget(buf, memory_budget).map(Self)
167    }
168}
169
170/// Decodes an RLP list, stopping once the cumulative [`InMemorySize`] of decoded items exceeds
171/// `memory_budget` bytes. Any remaining items in the payload are skipped.
172pub fn decode_list_with_memory_budget<T: Decodable + InMemorySize>(
173    buf: &mut &[u8],
174    memory_budget: usize,
175) -> alloy_rlp::Result<Vec<T>> {
176    let header = Header::decode(buf)?;
177    if !header.list {
178        return Err(alloy_rlp::Error::UnexpectedString);
179    }
180    if buf.len() < header.payload_length {
181        return Err(alloy_rlp::Error::InputTooShort);
182    }
183
184    let (payload, rest) = buf.split_at(header.payload_length);
185    let mut payload = payload;
186
187    let mut txs = Vec::with_capacity(estimated_transaction_list_capacity(header.payload_length));
188    let mut total_size = 0usize;
189
190    while !payload.is_empty() {
191        let item = T::decode(&mut payload)?;
192        total_size = total_size.saturating_add(item.size());
193
194        if total_size > memory_budget {
195            break;
196        }
197
198        txs.push(item);
199    }
200
201    *buf = rest;
202    Ok(txs)
203}
204
205// Keep this as a conservative hint: small lists stay allocation-free until the first push, while
206// large untrusted payloads cannot force an outsized preallocation.
207const MIN_TRANSACTION_RLP_SIZE_ESTIMATE: usize = 128;
208const MIN_PREALLOCATED_TRANSACTIONS: usize = 4;
209const MAX_PREALLOCATED_TRANSACTIONS: usize = 1024;
210
211const fn estimated_transaction_list_capacity(payload_length: usize) -> usize {
212    let estimate = payload_length / MIN_TRANSACTION_RLP_SIZE_ESTIMATE;
213    if estimate < MIN_PREALLOCATED_TRANSACTIONS {
214        0
215    } else if estimate > MAX_PREALLOCATED_TRANSACTIONS {
216        MAX_PREALLOCATED_TRANSACTIONS
217    } else {
218        estimate
219    }
220}
221
222/// Same as [`Transactions`] but this is intended as egress message send from local to _many_ peers.
223///
224/// The list of transactions is constructed on per-peers basis, but the underlying transaction
225/// objects are shared.
226#[derive(
227    Clone, Debug, PartialEq, Eq, RlpEncodableWrapper, RlpDecodableWrapper, Deref, IntoIterator,
228)]
229#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
230#[add_arbitrary_tests(rlp, 20)]
231pub struct SharedTransactions<T = TransactionSigned>(
232    /// New transactions for the peer to include in its mempool.
233    pub Vec<Arc<T>>,
234);
235
236/// A transaction that can be lazily encoded for pool-backed outbound propagation.
237pub trait BroadcastPoolTransaction:
238    Encodable + TxHashRef + Typed2718 + Send + Sync + 'static
239{
240}
241
242impl<T> BroadcastPoolTransaction for T where
243    T: Encodable + TxHashRef + Typed2718 + Send + Sync + 'static
244{
245}
246
247/// Shared cached encoding for an outbound transaction.
248///
249/// This keeps the transaction object and its encoded bytes behind shared references so cloned
250/// per-peer messages reuse the same EIP-2718 encoding.
251pub struct LazyEncoded<T: ?Sized> {
252    value: Arc<T>,
253    encoded: Arc<OnceLock<Bytes>>,
254}
255
256impl<T: ?Sized> Clone for LazyEncoded<T> {
257    fn clone(&self) -> Self {
258        Self { value: Arc::clone(&self.value), encoded: Arc::clone(&self.encoded) }
259    }
260}
261
262impl LazyEncoded<dyn BroadcastPoolTransaction> {
263    /// Wraps a transaction-like value and lazily caches its encoded bytes.
264    pub fn new<T>(value: T) -> Self
265    where
266        T: BroadcastPoolTransaction,
267    {
268        let value: Arc<dyn BroadcastPoolTransaction> = Arc::new(value);
269        Self { value, encoded: Arc::new(OnceLock::new()) }
270    }
271}
272
273impl<T: Encodable + ?Sized> Encodable for LazyEncoded<T> {
274    fn encode(&self, out: &mut dyn BufMut) {
275        let encoded = self.encoded.get_or_init(|| self.encode_uncached());
276        out.put_slice(encoded);
277    }
278
279    fn length(&self) -> usize {
280        self.encoded.get_or_init(|| self.encode_uncached()).len()
281    }
282}
283
284impl<T: Encodable + ?Sized> LazyEncoded<T> {
285    fn encode_uncached(&self) -> Bytes {
286        let mut out = Vec::with_capacity(self.value.length());
287        self.value.encode(&mut out);
288        out.into()
289    }
290}
291
292impl<T: TxHashRef + ?Sized> TxHashRef for LazyEncoded<T> {
293    fn tx_hash(&self) -> &TxHash {
294        self.value.tx_hash()
295    }
296}
297
298impl<T: Typed2718 + ?Sized> Typed2718 for LazyEncoded<T> {
299    fn ty(&self) -> u8 {
300        self.value.ty()
301    }
302}
303
304impl<T: ?Sized> Debug for LazyEncoded<T> {
305    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
306        f.debug_struct("LazyEncoded")
307            .field("is_cached", &self.encoded.get().is_some())
308            .finish_non_exhaustive()
309    }
310}
311
312/// A lazily encoded transaction used for pool-backed full transaction propagation.
313pub type LazyEncodedTransaction = LazyEncoded<dyn BroadcastPoolTransaction>;
314
315/// Outbound-only full transaction propagation message backed by pool transactions.
316///
317/// This encodes to the same `Transactions` wire payload as [`SharedTransactions`], but is used by
318/// the transaction manager when the source is the pool. Unlike [`SharedTransactions`], it can wrap
319/// pool transaction references directly and cache each transaction's encoded bytes across per-peer
320/// messages. Queued messages retain the pool-backed value and the shared cached bytes until they
321/// are sent.
322#[derive(Clone, Debug, Deref)]
323pub struct BroadcastPoolTransactions(pub Vec<LazyEncodedTransaction>);
324
325impl BroadcastPoolTransactions {
326    /// Returns an iterator over the transaction hashes.
327    pub fn iter_hashes(&self) -> impl Iterator<Item = &TxHash> + '_ {
328        self.0.iter().map(TxHashRef::tx_hash)
329    }
330}
331
332impl Encodable for BroadcastPoolTransactions {
333    fn encode(&self, out: &mut dyn BufMut) {
334        self.0.encode(out);
335    }
336
337    fn length(&self) -> usize {
338        self.0.length()
339    }
340}
341
342/// A wrapper type for all different new pooled transaction types
343#[derive(Clone, Debug, PartialEq, Eq)]
344#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
345pub enum NewPooledTransactionHashes {
346    /// A list of transaction hashes valid for [66-68)
347    Eth66(NewPooledTransactionHashes66),
348    /// A list of transaction hashes valid for [68-72)
349    ///
350    /// Note: it is assumed that the payload is valid (all vectors have the same length)
351    Eth68(NewPooledTransactionHashes68),
352    /// A list of transaction hashes valid from [72..]
353    ///
354    /// This extends the eth/68 announcement payload with the `cell_mask` field introduced by
355    /// [EIP-8070](https://eips.ethereum.org/EIPS/eip-8070).
356    ///
357    /// Note: it is assumed that the payload is valid (all vectors have the same length)
358    Eth72(NewPooledTransactionHashes72),
359}
360
361// === impl NewPooledTransactionHashes ===
362
363impl NewPooledTransactionHashes {
364    /// Returns the message [`EthVersion`].
365    pub const fn version(&self) -> EthVersion {
366        match self {
367            Self::Eth66(_) => EthVersion::Eth66,
368            Self::Eth68(_) => EthVersion::Eth68,
369            Self::Eth72(_) => EthVersion::Eth72,
370        }
371    }
372
373    /// Returns `true` if the payload is valid for the given version
374    pub const fn is_valid_for_version(&self, version: EthVersion) -> bool {
375        match self {
376            Self::Eth66(_) => {
377                matches!(version, EthVersion::Eth67 | EthVersion::Eth66)
378            }
379            Self::Eth68(_) => {
380                matches!(
381                    version,
382                    EthVersion::Eth68 | EthVersion::Eth69 | EthVersion::Eth70 | EthVersion::Eth71
383                )
384            }
385            Self::Eth72(_) => {
386                matches!(version, EthVersion::Eth72)
387            }
388        }
389    }
390
391    /// Returns an iterator over all transaction hashes.
392    pub fn iter_hashes(&self) -> impl Iterator<Item = &B256> + '_ {
393        match self {
394            Self::Eth66(msg) => msg.iter(),
395            Self::Eth68(msg) => msg.hashes.iter(),
396            Self::Eth72(msg) => msg.hashes.iter(),
397        }
398    }
399
400    /// Returns an immutable reference to transaction hashes.
401    pub const fn hashes(&self) -> &Vec<B256> {
402        match self {
403            Self::Eth66(msg) => &msg.0,
404            Self::Eth68(msg) => &msg.hashes,
405            Self::Eth72(msg) => &msg.hashes,
406        }
407    }
408
409    /// Returns a mutable reference to transaction hashes.
410    pub const fn hashes_mut(&mut self) -> &mut Vec<B256> {
411        match self {
412            Self::Eth66(msg) => &mut msg.0,
413            Self::Eth68(msg) => &mut msg.hashes,
414            Self::Eth72(msg) => &mut msg.hashes,
415        }
416    }
417
418    /// Consumes the type and returns all hashes
419    pub fn into_hashes(self) -> Vec<B256> {
420        match self {
421            Self::Eth66(msg) => msg.0,
422            Self::Eth68(msg) => msg.hashes,
423            Self::Eth72(msg) => msg.hashes,
424        }
425    }
426
427    /// Returns an iterator over all transaction hashes.
428    pub fn into_iter_hashes(self) -> impl Iterator<Item = B256> {
429        match self {
430            Self::Eth66(msg) => msg.into_iter(),
431            Self::Eth68(msg) => msg.hashes.into_iter(),
432            Self::Eth72(msg) => msg.hashes.into_iter(),
433        }
434    }
435
436    /// Shortens the number of hashes in the message, keeping the first `len` hashes and dropping
437    /// the rest. If `len` is greater than the number of hashes, this has no effect.
438    pub fn truncate(&mut self, len: usize) {
439        match self {
440            Self::Eth66(msg) => msg.truncate(len),
441            Self::Eth68(msg) => {
442                msg.types.truncate(len);
443                msg.sizes.truncate(len);
444                msg.hashes.truncate(len);
445            }
446            Self::Eth72(msg) => {
447                msg.types.truncate(len);
448                msg.sizes.truncate(len);
449                msg.hashes.truncate(len);
450            }
451        }
452    }
453
454    /// Returns true if the message is empty
455    pub const fn is_empty(&self) -> bool {
456        match self {
457            Self::Eth66(msg) => msg.0.is_empty(),
458            Self::Eth68(msg) => msg.hashes.is_empty(),
459            Self::Eth72(msg) => msg.hashes.is_empty(),
460        }
461    }
462
463    /// Returns the number of hashes in the message
464    pub const fn len(&self) -> usize {
465        match self {
466            Self::Eth66(msg) => msg.0.len(),
467            Self::Eth68(msg) => msg.hashes.len(),
468            Self::Eth72(msg) => msg.hashes.len(),
469        }
470    }
471
472    /// Returns an immutable reference to the inner type if this is an eth68 announcement.
473    pub const fn as_eth72(&self) -> Option<&NewPooledTransactionHashes72> {
474        match self {
475            Self::Eth66(_) | Self::Eth68(_) => None,
476            Self::Eth72(msg) => Some(msg),
477        }
478    }
479
480    /// Returns a mutable reference to the inner type if this is an eth68 announcement.
481    pub const fn as_eth72_mut(&mut self) -> Option<&mut NewPooledTransactionHashes72> {
482        match self {
483            Self::Eth66(_) | Self::Eth68(_) => None,
484            Self::Eth72(msg) => Some(msg),
485        }
486    }
487
488    /// Returns an immutable reference to the inner type if this is an eth68 announcement.
489    pub const fn as_eth68(&self) -> Option<&NewPooledTransactionHashes68> {
490        match self {
491            Self::Eth66(_) | Self::Eth72(_) => None,
492            Self::Eth68(msg) => Some(msg),
493        }
494    }
495
496    /// Returns a mutable reference to the inner type if this is an eth68 announcement.
497    pub const fn as_eth68_mut(&mut self) -> Option<&mut NewPooledTransactionHashes68> {
498        match self {
499            Self::Eth66(_) | Self::Eth72(_) => None,
500            Self::Eth68(msg) => Some(msg),
501        }
502    }
503
504    /// Returns a mutable reference to the inner type if this is an eth66 announcement.
505    pub const fn as_eth66_mut(&mut self) -> Option<&mut NewPooledTransactionHashes66> {
506        match self {
507            Self::Eth66(msg) => Some(msg),
508            Self::Eth68(_) | Self::Eth72(_) => None,
509        }
510    }
511
512    /// Returns the inner type if this is an eth68 announcement.
513    pub fn take_eth68(&mut self) -> Option<NewPooledTransactionHashes68> {
514        match self {
515            Self::Eth66(_) | Self::Eth72(_) => None,
516            Self::Eth68(msg) => Some(mem::take(msg)),
517        }
518    }
519
520    /// Returns the inner type if this is an eth66 announcement.
521    pub fn take_eth66(&mut self) -> Option<NewPooledTransactionHashes66> {
522        match self {
523            Self::Eth66(msg) => Some(mem::take(msg)),
524            Self::Eth68(_) | Self::Eth72(_) => None,
525        }
526    }
527}
528
529impl<N: NetworkPrimitives> From<NewPooledTransactionHashes> for EthMessage<N> {
530    fn from(value: NewPooledTransactionHashes) -> Self {
531        match value {
532            NewPooledTransactionHashes::Eth66(msg) => Self::NewPooledTransactionHashes66(msg),
533            NewPooledTransactionHashes::Eth68(msg) => Self::NewPooledTransactionHashes68(msg),
534            NewPooledTransactionHashes::Eth72(msg) => Self::NewPooledTransactionHashes72(msg),
535        }
536    }
537}
538
539impl From<NewPooledTransactionHashes66> for NewPooledTransactionHashes {
540    fn from(hashes: NewPooledTransactionHashes66) -> Self {
541        Self::Eth66(hashes)
542    }
543}
544
545impl From<NewPooledTransactionHashes68> for NewPooledTransactionHashes {
546    fn from(hashes: NewPooledTransactionHashes68) -> Self {
547        Self::Eth68(hashes)
548    }
549}
550
551impl From<NewPooledTransactionHashes72> for NewPooledTransactionHashes {
552    fn from(hashes: NewPooledTransactionHashes72) -> Self {
553        Self::Eth72(hashes)
554    }
555}
556
557/// This informs peers of transaction hashes for transactions that have appeared on the network,
558/// but have not been included in a block.
559#[derive(
560    Clone,
561    Debug,
562    PartialEq,
563    Eq,
564    RlpEncodableWrapper,
565    RlpDecodableWrapper,
566    Default,
567    Deref,
568    DerefMut,
569    IntoIterator,
570)]
571#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
572#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
573#[add_arbitrary_tests(rlp)]
574pub struct NewPooledTransactionHashes66(
575    /// Transaction hashes for new transactions that have appeared on the network.
576    /// Clients should request the transactions with the given hashes using a
577    /// [`GetPooledTransactions`](crate::GetPooledTransactions) message.
578    pub Vec<B256>,
579);
580
581impl NewPooledTransactionHashes66 {
582    /// Returns a new instance with capacity for `capacity` hashes.
583    pub fn with_capacity(capacity: usize) -> Self {
584        Self(Vec::with_capacity(capacity))
585    }
586}
587
588impl From<Vec<B256>> for NewPooledTransactionHashes66 {
589    fn from(v: Vec<B256>) -> Self {
590        Self(v)
591    }
592}
593
594/// Same as [`NewPooledTransactionHashes66`] but extends that beside the transaction hashes,
595/// the node sends the transaction types and their sizes (as defined in EIP-2718) as well.
596#[derive(Clone, Debug, PartialEq, Eq, Default)]
597#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
598pub struct NewPooledTransactionHashes68 {
599    /// Transaction types for new transactions that have appeared on the network.
600    ///
601    /// ## Note on RLP encoding and decoding
602    ///
603    /// In the [eth/68 spec](https://eips.ethereum.org/EIPS/eip-5793#specification) this is defined
604    /// the following way:
605    ///  * `[type_0: B_1, type_1: B_1, ...]`
606    ///
607    /// This would make it seem like the [`Encodable`] and
608    /// [`Decodable`] implementations should directly use a `Vec<u8>` for
609    /// encoding and decoding, because it looks like this field should be encoded as a _list_ of
610    /// bytes.
611    ///
612    /// However, [this is implemented in geth as a `[]byte`
613    /// type](https://github.com/ethereum/go-ethereum/blob/82d934b1dd80cdd8190803ea9f73ed2c345e2576/eth/protocols/eth/protocol.go#L308-L313),
614    /// which [ends up being encoded as a RLP
615    /// string](https://github.com/ethereum/go-ethereum/blob/82d934b1dd80cdd8190803ea9f73ed2c345e2576/rlp/encode_test.go#L171-L176),
616    /// **not** a RLP list.
617    ///
618    /// Because of this, we do not directly use the `Vec<u8>` when encoding and decoding, and
619    /// instead use the [`Encodable`] and [`Decodable`]
620    /// implementations for `&[u8]` instead, which encodes into a RLP string, and expects an RLP
621    /// string when decoding.
622    pub types: Vec<u8>,
623    /// Transaction sizes for new transactions that have appeared on the network.
624    pub sizes: Vec<usize>,
625    /// Transaction hashes for new transactions that have appeared on the network.
626    pub hashes: Vec<B256>,
627}
628
629#[cfg(feature = "arbitrary")]
630impl proptest::prelude::Arbitrary for NewPooledTransactionHashes68 {
631    type Parameters = ();
632    fn arbitrary_with(_args: ()) -> Self::Strategy {
633        use proptest::{collection::vec, prelude::*};
634        // Generate a single random length for all vectors
635        let vec_length = any::<usize>().prop_map(|x| x % 100 + 1); // Lengths between 1 and 100
636
637        vec_length
638            .prop_flat_map(|len| {
639                // Use the generated length to create vectors of TxType, usize, and B256
640                let types_vec = vec(
641                    proptest_arbitrary_interop::arb::<reth_ethereum_primitives::TxType>()
642                        .prop_map(|ty| ty as u8),
643                    len..=len,
644                );
645
646                // Map the usize values to the range 0..131072(0x20000)
647                let sizes_vec = vec(proptest::num::usize::ANY.prop_map(|x| x % 131072), len..=len);
648                let hashes_vec = vec(any::<B256>(), len..=len);
649
650                (types_vec, sizes_vec, hashes_vec)
651            })
652            .prop_map(|(types, sizes, hashes)| Self { types, sizes, hashes })
653            .boxed()
654    }
655
656    type Strategy = proptest::prelude::BoxedStrategy<Self>;
657}
658
659impl NewPooledTransactionHashes68 {
660    /// Returns a new instance with capacity for `capacity` entries.
661    pub fn with_capacity(capacity: usize) -> Self {
662        Self {
663            types: Vec::with_capacity(capacity),
664            sizes: Vec::with_capacity(capacity),
665            hashes: Vec::with_capacity(capacity),
666        }
667    }
668
669    /// Returns an iterator over tx hashes zipped with corresponding metadata.
670    pub fn metadata_iter(&self) -> impl Iterator<Item = (&B256, (u8, usize))> {
671        self.hashes.iter().zip(self.types.iter().copied().zip(self.sizes.iter().copied()))
672    }
673
674    /// Appends a transaction
675    pub fn push<T: SignedTransaction>(&mut self, tx: &T) {
676        self.hashes.push(*tx.tx_hash());
677        self.sizes.push(tx.encode_2718_len());
678        self.types.push(tx.ty());
679    }
680
681    /// Appends the provided transactions
682    pub fn extend<'a, T: SignedTransaction>(&mut self, txs: impl IntoIterator<Item = &'a T>) {
683        for tx in txs {
684            self.push(tx);
685        }
686    }
687
688    /// Shrinks the capacity of the message vectors as much as possible.
689    pub fn shrink_to_fit(&mut self) {
690        self.hashes.shrink_to_fit();
691        self.sizes.shrink_to_fit();
692        self.types.shrink_to_fit()
693    }
694
695    /// Consumes and appends a transaction
696    pub fn with_transaction<T: SignedTransaction>(mut self, tx: &T) -> Self {
697        self.push(tx);
698        self
699    }
700
701    /// Consumes and appends the provided transactions
702    pub fn with_transactions<'a, T: SignedTransaction>(
703        mut self,
704        txs: impl IntoIterator<Item = &'a T>,
705    ) -> Self {
706        self.extend(txs);
707        self
708    }
709}
710
711impl Encodable for NewPooledTransactionHashes68 {
712    fn encode(&self, out: &mut dyn bytes::BufMut) {
713        #[derive(RlpEncodable)]
714        struct EncodableNewPooledTransactionHashes68<'a> {
715            types: &'a [u8],
716            sizes: &'a Vec<usize>,
717            hashes: &'a Vec<B256>,
718        }
719
720        let encodable = EncodableNewPooledTransactionHashes68 {
721            types: &self.types[..],
722            sizes: &self.sizes,
723            hashes: &self.hashes,
724        };
725
726        encodable.encode(out);
727    }
728    fn length(&self) -> usize {
729        #[derive(RlpEncodable)]
730        struct EncodableNewPooledTransactionHashes68<'a> {
731            types: &'a [u8],
732            sizes: &'a Vec<usize>,
733            hashes: &'a Vec<B256>,
734        }
735
736        let encodable = EncodableNewPooledTransactionHashes68 {
737            types: &self.types[..],
738            sizes: &self.sizes,
739            hashes: &self.hashes,
740        };
741
742        encodable.length()
743    }
744}
745
746impl Decodable for NewPooledTransactionHashes68 {
747    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
748        let Header { list, payload_length } = Header::decode(buf)?;
749        if !list {
750            return Err(alloy_rlp::Error::UnexpectedString)
751        }
752        if buf.len() < payload_length {
753            return Err(alloy_rlp::Error::InputTooShort)
754        }
755
756        let (mut payload, rest) = buf.split_at(payload_length);
757        let (types, sizes, hashes) = decode_pooled_transaction_hashes_payload(&mut payload)?;
758
759        if !payload.is_empty() {
760            return Err(alloy_rlp::Error::ListLengthMismatch {
761                expected: payload_length,
762                got: payload_length - payload.len(),
763            })
764        }
765
766        ensure_pooled_transaction_hashes_lengths(hashes.len(), types.len(), sizes.len())?;
767
768        let msg = Self { types, sizes, hashes };
769
770        *buf = rest;
771        Ok(msg)
772    }
773}
774
775/// Same as [`NewPooledTransactionHashes68`] but adds the eth/72 `cell_mask` field from
776/// [EIP-8070](https://eips.ethereum.org/EIPS/eip-8070).
777#[derive(Clone, Debug, PartialEq, Eq, Default)]
778#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
779pub struct NewPooledTransactionHashes72 {
780    /// Transaction types for new transactions that have appeared on the network.
781    ///
782    /// ## Note on RLP encoding and decoding
783    ///
784    /// In the [eth/72 spec](https://eips.ethereum.org/EIPS/eip-8070#specification) this is defined
785    /// the following way:
786    ///  * `[types: B, [size_0: P, size_1: P, ...], [hash_0: B_32, hash_1: B_32, ...], cell_mask:
787    ///    B_16]`
788    pub types: Vec<u8>,
789    /// Transaction sizes for new transactions that have appeared on the network.
790    pub sizes: Vec<usize>,
791    /// Transaction hashes for new transactions that have appeared on the network.
792    pub hashes: Vec<B256>,
793    /// Cell availability mask for type 3 (blob) transactions announced by this message.
794    ///
795    /// Per [EIP-8070](https://eips.ethereum.org/EIPS/eip-8070), this is a `B_16`
796    /// bitarray over `CELLS_PER_EXT_BLOB`; bit `i` is set when the announcer has column
797    /// `i` available for every type 3 transaction in the message. `None` encodes as RLP
798    /// `nil` and must be used when no type 3 transactions are announced.
799    pub cell_mask: Option<B128>,
800}
801
802#[cfg(feature = "arbitrary")]
803impl proptest::prelude::Arbitrary for NewPooledTransactionHashes72 {
804    type Parameters = ();
805    fn arbitrary_with(_args: ()) -> Self::Strategy {
806        use proptest::{collection::vec, prelude::*};
807        // Generate a single random length for all vectors
808        let vec_length = any::<usize>().prop_map(|x| x % 100 + 1); // Lengths between 1 and 100
809
810        vec_length
811            .prop_flat_map(|len| {
812                // Use the generated length to create vectors of TxType, usize, and B256
813                let types_vec = vec(
814                    proptest_arbitrary_interop::arb::<reth_ethereum_primitives::TxType>()
815                        .prop_map(|ty| ty as u8),
816                    len..=len,
817                );
818
819                // Map the usize values to the range 0..131072(0x20000)
820                let sizes_vec = vec(proptest::num::usize::ANY.prop_map(|x| x % 131072), len..=len);
821                let hashes_vec = vec(any::<B256>(), len..=len);
822                let cell_mask = any::<Option<B128>>();
823
824                (types_vec, sizes_vec, hashes_vec, cell_mask)
825            })
826            .prop_map(|(types, sizes, hashes, cell_mask)| Self { types, sizes, hashes, cell_mask })
827            .boxed()
828    }
829
830    type Strategy = proptest::prelude::BoxedStrategy<Self>;
831}
832
833impl NewPooledTransactionHashes72 {
834    /// Returns a new instance with capacity for `capacity` entries and no cell mask.
835    pub fn with_capacity(capacity: usize) -> Self {
836        Self {
837            types: Vec::with_capacity(capacity),
838            sizes: Vec::with_capacity(capacity),
839            hashes: Vec::with_capacity(capacity),
840            cell_mask: None,
841        }
842    }
843
844    /// Returns an iterator over tx hashes zipped with corresponding metadata.
845    pub fn metadata_iter(&self) -> impl Iterator<Item = (&B256, (u8, usize))> {
846        self.hashes.iter().zip(self.types.iter().copied().zip(self.sizes.iter().copied()))
847    }
848
849    /// Appends a transaction
850    pub fn push<T: SignedTransaction>(&mut self, tx: &T) {
851        self.hashes.push(*tx.tx_hash());
852        self.sizes.push(tx.encode_2718_len());
853        self.types.push(tx.ty());
854    }
855
856    /// Appends the provided transactions
857    pub fn extend<'a, T: SignedTransaction>(&mut self, txs: impl IntoIterator<Item = &'a T>) {
858        for tx in txs {
859            self.push(tx);
860        }
861    }
862
863    /// Shrinks the capacity of the message vectors as much as possible.
864    pub fn shrink_to_fit(&mut self) {
865        self.hashes.shrink_to_fit();
866        self.sizes.shrink_to_fit();
867        self.types.shrink_to_fit()
868    }
869
870    /// Consumes and appends a transaction
871    pub fn with_transaction<T: SignedTransaction>(mut self, tx: &T) -> Self {
872        self.push(tx);
873        self
874    }
875
876    /// Consumes and appends the provided transactions
877    pub fn with_transactions<'a, T: SignedTransaction>(
878        mut self,
879        txs: impl IntoIterator<Item = &'a T>,
880    ) -> Self {
881        self.extend(txs);
882        self
883    }
884
885    fn payload_length(&self) -> usize {
886        self.types.as_slice().length() +
887            self.sizes.length() +
888            self.hashes.length() +
889            self.cell_mask.as_ref().map_or(1, Encodable::length)
890    }
891}
892
893impl Encodable for NewPooledTransactionHashes72 {
894    fn encode(&self, out: &mut dyn bytes::BufMut) {
895        Header { list: true, payload_length: self.payload_length() }.encode(out);
896        self.types.as_slice().encode(out);
897        self.sizes.encode(out);
898        self.hashes.encode(out);
899        if let Some(cell_mask) = &self.cell_mask {
900            cell_mask.encode(out);
901        } else {
902            out.put_u8(alloy_rlp::EMPTY_STRING_CODE);
903        }
904    }
905
906    fn length(&self) -> usize {
907        Header { list: true, payload_length: self.payload_length() }.length_with_payload()
908    }
909}
910
911impl Decodable for NewPooledTransactionHashes72 {
912    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
913        let Header { list, payload_length } = Header::decode(buf)?;
914        if !list {
915            return Err(alloy_rlp::Error::UnexpectedString)
916        }
917        if buf.len() < payload_length {
918            return Err(alloy_rlp::Error::InputTooShort)
919        }
920
921        let (mut payload, rest) = buf.split_at(payload_length);
922        let (types, sizes, hashes) = decode_pooled_transaction_hashes_payload(&mut payload)?;
923        let Some(first_byte) = payload.first().copied() else {
924            return Err(alloy_rlp::Error::InputTooShort)
925        };
926        let cell_mask = if first_byte == alloy_rlp::EMPTY_STRING_CODE {
927            payload = &payload[1..];
928            None
929        } else {
930            Some(B128::decode(&mut payload)?)
931        };
932
933        if !payload.is_empty() {
934            return Err(alloy_rlp::Error::ListLengthMismatch {
935                expected: payload_length,
936                got: payload_length - payload.len(),
937            })
938        }
939
940        ensure_pooled_transaction_hashes_lengths(hashes.len(), types.len(), sizes.len())?;
941
942        *buf = rest;
943
944        Ok(Self { types, sizes, hashes, cell_mask })
945    }
946}
947
948/// Twice the spec'd soft limit for `NewPooledTransactionHashes` announcements.
949///
950/// This keeps capacity hints bounded when a malformed packet spends most of its bytes on the
951/// one-byte `types` string before the size and hash lists are validated.
952const NEW_POOLED_TRANSACTION_HASHES_DECODE_CAP: usize =
953    2 * SOFT_LIMIT_COUNT_HASHES_IN_NEW_POOLED_TRANSACTIONS_BROADCAST_MESSAGE;
954
955#[inline]
956fn decode_pooled_transaction_hashes_payload(
957    payload: &mut &[u8],
958) -> alloy_rlp::Result<(Vec<u8>, Vec<usize>, Vec<B256>)> {
959    let types = Bytes::decode(payload)?;
960    let capacity = types.len().min(NEW_POOLED_TRANSACTION_HASHES_DECODE_CAP);
961
962    let mut sizes = Vec::with_capacity(capacity);
963    decode_append(payload, &mut sizes)?;
964
965    let mut hashes = Vec::with_capacity(capacity);
966    decode_append(payload, &mut hashes)?;
967
968    Ok((types.into(), sizes, hashes))
969}
970
971#[inline]
972const fn ensure_pooled_transaction_hashes_lengths(
973    hashes_len: usize,
974    types_len: usize,
975    sizes_len: usize,
976) -> alloy_rlp::Result<()> {
977    if hashes_len != types_len {
978        return Err(alloy_rlp::Error::ListLengthMismatch { expected: hashes_len, got: types_len })
979    }
980    if hashes_len != sizes_len {
981        return Err(alloy_rlp::Error::ListLengthMismatch { expected: hashes_len, got: sizes_len })
982    }
983
984    Ok(())
985}
986
987/// Validation pass that checks for unique transaction hashes.
988pub trait DedupPayload {
989    /// Value type in [`PartiallyValidData`] map.
990    type Value;
991
992    /// The payload contains no entries.
993    fn is_empty(&self) -> bool;
994
995    /// Returns the number of entries.
996    fn len(&self) -> usize;
997
998    /// Consumes self, returning an iterator over hashes in payload.
999    fn dedup(self) -> PartiallyValidData<Self::Value>;
1000}
1001
1002/// Value in [`PartiallyValidData`] map obtained from an announcement.
1003pub type Eth68TxMetadata = Option<(u8, usize)>;
1004
1005impl DedupPayload for NewPooledTransactionHashes {
1006    type Value = Eth68TxMetadata;
1007
1008    fn is_empty(&self) -> bool {
1009        self.is_empty()
1010    }
1011
1012    fn len(&self) -> usize {
1013        self.len()
1014    }
1015
1016    fn dedup(self) -> PartiallyValidData<Self::Value> {
1017        match self {
1018            Self::Eth66(msg) => msg.dedup(),
1019            Self::Eth68(msg) => msg.dedup(),
1020            Self::Eth72(msg) => msg.dedup(),
1021        }
1022    }
1023}
1024
1025impl DedupPayload for NewPooledTransactionHashes72 {
1026    type Value = Eth68TxMetadata;
1027
1028    fn is_empty(&self) -> bool {
1029        self.hashes.is_empty()
1030    }
1031
1032    fn len(&self) -> usize {
1033        self.hashes.len()
1034    }
1035
1036    fn dedup(self) -> PartiallyValidData<Self::Value> {
1037        let Self { hashes, mut sizes, mut types, cell_mask } = self;
1038
1039        let mut deduped_data = B256Map::with_capacity_and_hasher(hashes.len(), Default::default());
1040
1041        for hash in hashes.into_iter().rev() {
1042            if let (Some(ty), Some(size)) = (types.pop(), sizes.pop()) {
1043                deduped_data.insert(hash, Some((ty, size)));
1044            }
1045        }
1046
1047        PartiallyValidData::from_raw_data_eth72_with_cell_mask(deduped_data, cell_mask)
1048    }
1049}
1050
1051impl DedupPayload for NewPooledTransactionHashes68 {
1052    type Value = Eth68TxMetadata;
1053
1054    fn is_empty(&self) -> bool {
1055        self.hashes.is_empty()
1056    }
1057
1058    fn len(&self) -> usize {
1059        self.hashes.len()
1060    }
1061
1062    fn dedup(self) -> PartiallyValidData<Self::Value> {
1063        let Self { hashes, mut sizes, mut types } = self;
1064
1065        let mut deduped_data = B256Map::with_capacity_and_hasher(hashes.len(), Default::default());
1066
1067        for hash in hashes.into_iter().rev() {
1068            if let (Some(ty), Some(size)) = (types.pop(), sizes.pop()) {
1069                deduped_data.insert(hash, Some((ty, size)));
1070            }
1071        }
1072
1073        PartiallyValidData::from_raw_data_eth68(deduped_data)
1074    }
1075}
1076
1077impl DedupPayload for NewPooledTransactionHashes66 {
1078    type Value = Eth68TxMetadata;
1079
1080    fn is_empty(&self) -> bool {
1081        self.0.is_empty()
1082    }
1083
1084    fn len(&self) -> usize {
1085        self.0.len()
1086    }
1087
1088    fn dedup(self) -> PartiallyValidData<Self::Value> {
1089        let Self(hashes) = self;
1090
1091        let mut deduped_data = B256Map::with_capacity_and_hasher(hashes.len(), Default::default());
1092
1093        let noop_value: Eth68TxMetadata = None;
1094
1095        for hash in hashes.into_iter().rev() {
1096            deduped_data.insert(hash, noop_value);
1097        }
1098
1099        PartiallyValidData::from_raw_data_eth66(deduped_data)
1100    }
1101}
1102
1103/// Interface for handling mempool message data. Used in various filters in pipelines in
1104/// `TransactionsManager` and in queries to `TransactionPool`.
1105pub trait HandleMempoolData {
1106    /// The announcement contains no entries.
1107    fn is_empty(&self) -> bool;
1108
1109    /// Returns the number of entries.
1110    fn len(&self) -> usize;
1111
1112    /// Retain only entries for which the hash in the entry satisfies a given predicate.
1113    fn retain_by_hash(&mut self, f: impl FnMut(&TxHash) -> bool);
1114}
1115
1116/// Extension of [`HandleMempoolData`] interface, for mempool messages that are versioned.
1117pub trait HandleVersionedMempoolData {
1118    /// Returns the announcement version, either [`Eth66`](EthVersion::Eth66) or
1119    /// [`Eth68`](EthVersion::Eth68).
1120    fn msg_version(&self) -> EthVersion;
1121}
1122
1123impl<T: SignedTransaction> HandleMempoolData for Vec<T> {
1124    fn is_empty(&self) -> bool {
1125        self.is_empty()
1126    }
1127
1128    fn len(&self) -> usize {
1129        self.len()
1130    }
1131
1132    fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
1133        self.retain(|tx| f(tx.tx_hash()))
1134    }
1135}
1136
1137macro_rules! handle_mempool_data_map_impl {
1138    ($data_ty:ty, $(<$generic:ident>)?) => {
1139        impl$(<$generic>)? HandleMempoolData for $data_ty {
1140            fn is_empty(&self) -> bool {
1141                self.data.is_empty()
1142            }
1143
1144            fn len(&self) -> usize {
1145                self.data.len()
1146            }
1147
1148            fn retain_by_hash(&mut self, mut f: impl FnMut(&TxHash) -> bool) {
1149                self.data.retain(|hash, _| f(hash));
1150            }
1151        }
1152    };
1153}
1154
1155/// Data that has passed an initial validation pass that is not specific to any mempool message
1156/// type.
1157#[derive(Debug, Deref, DerefMut, IntoIterator)]
1158pub struct PartiallyValidData<V> {
1159    #[deref]
1160    #[deref_mut]
1161    #[into_iterator]
1162    data: B256Map<V>,
1163    version: Option<EthVersion>,
1164    /// The eth/72 message-level cell mask, if present.
1165    cell_mask: Option<B128>,
1166}
1167
1168handle_mempool_data_map_impl!(PartiallyValidData<V>, <V>);
1169
1170impl<V> PartiallyValidData<V> {
1171    /// Wraps raw data.
1172    pub const fn from_raw_data(data: B256Map<V>, version: Option<EthVersion>) -> Self {
1173        Self { data, version, cell_mask: None }
1174    }
1175
1176    /// Wraps raw data with version [`EthVersion::Eth72`].
1177    pub const fn from_raw_data_eth72(data: B256Map<V>) -> Self {
1178        Self::from_raw_data(data, Some(EthVersion::Eth72))
1179    }
1180
1181    /// Wraps raw data with an eth/72 message-level cell mask.
1182    pub const fn from_raw_data_eth72_with_cell_mask(
1183        data: B256Map<V>,
1184        cell_mask: Option<B128>,
1185    ) -> Self {
1186        Self { data, version: Some(EthVersion::Eth72), cell_mask }
1187    }
1188
1189    /// Wraps raw data with version [`EthVersion::Eth68`].
1190    pub const fn from_raw_data_eth68(data: B256Map<V>) -> Self {
1191        Self::from_raw_data(data, Some(EthVersion::Eth68))
1192    }
1193
1194    /// Wraps raw data with version [`EthVersion::Eth66`].
1195    pub const fn from_raw_data_eth66(data: B256Map<V>) -> Self {
1196        Self::from_raw_data(data, Some(EthVersion::Eth66))
1197    }
1198
1199    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth72`](EthVersion::Eth72)
1200    /// announcement.
1201    pub fn empty_eth72() -> Self {
1202        Self::from_raw_data_eth72(B256Map::default())
1203    }
1204
1205    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth68`](EthVersion::Eth68)
1206    /// announcement.
1207    pub fn empty_eth68() -> Self {
1208        Self::from_raw_data_eth68(B256Map::default())
1209    }
1210
1211    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth66`](EthVersion::Eth66)
1212    /// announcement.
1213    pub fn empty_eth66() -> Self {
1214        Self::from_raw_data_eth66(B256Map::default())
1215    }
1216
1217    /// Returns the version of the message this data was received in if different versions of the
1218    /// message exist.
1219    pub const fn msg_version(&self) -> Option<EthVersion> {
1220        self.version
1221    }
1222
1223    /// Returns the eth/72 message-level cell mask, if present.
1224    pub const fn eth72_cell_mask(&self) -> Option<B128> {
1225        self.cell_mask
1226    }
1227
1228    /// Destructs returning the validated data.
1229    pub fn into_data(self) -> B256Map<V> {
1230        self.data
1231    }
1232}
1233
1234/// Partially validated data from an announcement or a
1235/// [`PooledTransactions`](crate::PooledTransactions) response.
1236#[derive(Debug, Deref, DerefMut, IntoIterator, From)]
1237pub struct ValidAnnouncementData {
1238    #[deref]
1239    #[deref_mut]
1240    #[into_iterator]
1241    data: B256Map<Eth68TxMetadata>,
1242    version: EthVersion,
1243    /// The eth/72 message-level cell mask, if present.
1244    cell_mask: Option<B128>,
1245}
1246
1247handle_mempool_data_map_impl!(ValidAnnouncementData,);
1248
1249impl ValidAnnouncementData {
1250    /// Destructs returning only the valid hashes and the announcement message version. Caution! If
1251    /// this is [`Eth68`](EthVersion::Eth68) announcement data, this drops the metadata.
1252    pub fn into_request_hashes(self) -> (RequestTxHashes, EthVersion) {
1253        let hashes = self.data.into_keys().collect::<B256Set>();
1254
1255        (RequestTxHashes::new(hashes), self.version)
1256    }
1257
1258    /// Conversion from [`PartiallyValidData`] from an announcement. Note! [`PartiallyValidData`]
1259    /// from an announcement, should have some [`EthVersion`]. Panics if [`PartiallyValidData`] has
1260    /// version set to `None`.
1261    pub fn from_partially_valid_data(data: PartiallyValidData<Eth68TxMetadata>) -> Self {
1262        let PartiallyValidData { data, version, cell_mask } = data;
1263
1264        let version = version.expect("should have eth version for conversion");
1265
1266        Self { data, version, cell_mask }
1267    }
1268
1269    /// Returns the eth/72 message-level cell mask, if present.
1270    pub const fn eth72_cell_mask(&self) -> Option<B128> {
1271        self.cell_mask
1272    }
1273
1274    /// Destructs returning the validated data.
1275    pub fn into_data(self) -> B256Map<Eth68TxMetadata> {
1276        self.data
1277    }
1278}
1279
1280impl HandleVersionedMempoolData for ValidAnnouncementData {
1281    fn msg_version(&self) -> EthVersion {
1282        self.version
1283    }
1284}
1285
1286/// Hashes to request from a peer.
1287#[derive(Debug, Default, Deref, DerefMut, IntoIterator, Constructor)]
1288pub struct RequestTxHashes {
1289    #[deref]
1290    #[deref_mut]
1291    #[into_iterator(owned, ref)]
1292    hashes: B256Set,
1293}
1294
1295impl RequestTxHashes {
1296    /// Returns a new [`RequestTxHashes`] with given capacity for hashes. Caution! Make sure to
1297    /// call `shrink_to_fit` on [`RequestTxHashes`] when full, especially where it will
1298    /// be stored in its entirety like in the future waiting for a
1299    /// [`GetPooledTransactions`](crate::GetPooledTransactions) request to resolve.
1300    pub fn with_capacity(capacity: usize) -> Self {
1301        Self::new(B256Set::with_capacity_and_hasher(capacity, Default::default()))
1302    }
1303
1304    /// Returns a new empty instance.
1305    fn empty() -> Self {
1306        Self::new(B256Set::default())
1307    }
1308
1309    /// Retains the given number of elements, returning an iterator over the rest.
1310    pub fn retain_count(&mut self, count: usize) -> Self {
1311        let rest_capacity = self.hashes.len().saturating_sub(count);
1312        if rest_capacity == 0 {
1313            return Self::empty()
1314        }
1315        let mut rest = Self::with_capacity(rest_capacity);
1316
1317        let mut i = 0;
1318        self.hashes.retain(|hash| {
1319            if i >= count {
1320                rest.insert(*hash);
1321                return false
1322            }
1323            i += 1;
1324
1325            true
1326        });
1327
1328        rest
1329    }
1330}
1331
1332impl FromIterator<(TxHash, Eth68TxMetadata)> for RequestTxHashes {
1333    fn from_iter<I: IntoIterator<Item = (TxHash, Eth68TxMetadata)>>(iter: I) -> Self {
1334        Self::new(iter.into_iter().map(|(hash, _)| hash).collect())
1335    }
1336}
1337
1338/// The earliest block, the latest block and hash of the latest block which can be provided.
1339/// See [BlockRangeUpdate](https://github.com/ethereum/devp2p/blob/master/caps/eth.md#blockrangeupdate-0x11).
1340#[derive(Clone, Debug, PartialEq, Eq, Default, RlpEncodable, RlpDecodable)]
1341#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
1342#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))]
1343pub struct BlockRangeUpdate {
1344    /// The earliest block which is available.
1345    pub earliest: u64,
1346    /// The latest block which is available.
1347    pub latest: u64,
1348    /// Latest available block's hash.
1349    pub latest_hash: B256,
1350}
1351
1352impl InMemorySize for NewPooledTransactionHashes {
1353    fn size(&self) -> usize {
1354        match self {
1355            Self::Eth66(msg) => msg.0.len() * core::mem::size_of::<B256>(),
1356            Self::Eth68(msg) => {
1357                msg.types.len() * core::mem::size_of::<u8>() +
1358                    msg.sizes.len() * core::mem::size_of::<usize>() +
1359                    msg.hashes.len() * core::mem::size_of::<B256>()
1360            }
1361            Self::Eth72(msg) => {
1362                msg.types.len() * core::mem::size_of::<u8>() +
1363                    msg.sizes.len() * core::mem::size_of::<usize>() +
1364                    msg.hashes.len() * core::mem::size_of::<B256>() +
1365                    core::mem::size_of::<B128>()
1366            }
1367        }
1368    }
1369}
1370
1371#[cfg(test)]
1372mod tests {
1373    use super::*;
1374    use alloy_consensus::{transaction::TxHashRef, Typed2718};
1375    use alloy_eips::eip2718::Encodable2718;
1376    use alloy_primitives::{b256, hex, Bytes, Signature, U256};
1377    use alloy_rlp::{RlpDecodable, RlpEncodable};
1378    use proptest::prelude::*;
1379    use reth_ethereum_primitives::{Transaction, TransactionSigned};
1380    use std::str::FromStr;
1381
1382    /// Takes as input a struct / encoded hex message pair, ensuring that we encode to the exact hex
1383    /// message, and decode to the exact struct.
1384    fn test_encoding_vector<T: Encodable + Decodable + PartialEq + core::fmt::Debug>(
1385        input: (T, &[u8]),
1386    ) {
1387        let (expected_decoded, expected_encoded) = input;
1388        let mut encoded = Vec::new();
1389        expected_decoded.encode(&mut encoded);
1390
1391        assert_eq!(hex::encode(&encoded), hex::encode(expected_encoded));
1392
1393        let decoded = T::decode(&mut encoded.as_ref()).unwrap();
1394        assert_eq!(expected_decoded, decoded);
1395    }
1396
1397    fn encoded<T: Encodable>(value: &T) -> Vec<u8> {
1398        let mut out = Vec::new();
1399        value.encode(&mut out);
1400        out
1401    }
1402
1403    #[derive(RlpEncodable, RlpDecodable)]
1404    struct EncodableNewPooledTransactionHashes68 {
1405        types: Bytes,
1406        sizes: Vec<usize>,
1407        hashes: Vec<B256>,
1408    }
1409
1410    type NewPooledTransactionHashes68Fields = (Vec<u8>, Vec<usize>, Vec<B256>);
1411
1412    fn decode_eth68_hashes_derived(
1413        buf: &mut &[u8],
1414    ) -> alloy_rlp::Result<NewPooledTransactionHashes68> {
1415        let encodable = EncodableNewPooledTransactionHashes68::decode(buf)?;
1416        let msg = NewPooledTransactionHashes68 {
1417            types: encodable.types.into(),
1418            sizes: encodable.sizes,
1419            hashes: encodable.hashes,
1420        };
1421
1422        ensure_pooled_transaction_hashes_lengths(
1423            msg.hashes.len(),
1424            msg.types.len(),
1425            msg.sizes.len(),
1426        )?;
1427
1428        Ok(msg)
1429    }
1430
1431    fn eth68_hash_fields_strategy() -> impl Strategy<Value = NewPooledTransactionHashes68Fields> {
1432        (0usize..128, 0usize..128, 0usize..128).prop_flat_map(
1433            |(types_len, sizes_len, hashes_len)| {
1434                (
1435                    proptest::collection::vec(any::<u8>(), types_len),
1436                    proptest::collection::vec(0usize..131_072, sizes_len),
1437                    proptest::collection::vec(any::<B256>(), hashes_len),
1438                )
1439            },
1440        )
1441    }
1442
1443    proptest! {
1444        #[test]
1445        fn broadcast_pool_transactions_match_shared_transactions_encoding(
1446            txs in proptest::collection::vec(
1447                proptest_arbitrary_interop::arb::<TransactionSigned>(),
1448                0..32,
1449            )
1450        ) {
1451            let shared = SharedTransactions::<TransactionSigned>(
1452                txs.iter().cloned().map(Arc::new).collect(),
1453            );
1454            let broadcast = BroadcastPoolTransactions(
1455                txs.iter().cloned().map(LazyEncoded::new).collect(),
1456            );
1457
1458            prop_assert_eq!(broadcast.length(), shared.length());
1459
1460            let shared_encoded = encoded(&shared);
1461            let broadcast_encoded = encoded(&broadcast);
1462            prop_assert_eq!(&broadcast_encoded, &shared_encoded);
1463
1464            let broadcast_encoded_cached = encoded(&broadcast);
1465            prop_assert_eq!(&broadcast_encoded_cached, &shared_encoded);
1466
1467            let mut shared_bytes = shared_encoded.as_slice();
1468            let decoded_shared = SharedTransactions::<TransactionSigned>::decode(&mut shared_bytes)
1469                .expect("shared transactions decode");
1470            prop_assert!(shared_bytes.is_empty());
1471
1472            let mut broadcast_bytes = broadcast_encoded.as_slice();
1473            let decoded_broadcast =
1474                SharedTransactions::<TransactionSigned>::decode(&mut broadcast_bytes)
1475                    .expect("broadcast pool transactions decode as shared transactions");
1476            prop_assert!(broadcast_bytes.is_empty());
1477
1478            prop_assert_eq!(decoded_broadcast, decoded_shared);
1479        }
1480
1481        #[test]
1482        fn eth_68_handrolled_decode_matches_derived_implementation(
1483            (types, sizes, hashes) in eth68_hash_fields_strategy()
1484        ) {
1485            let encodable = EncodableNewPooledTransactionHashes68 {
1486                types: Bytes::from(types),
1487                sizes,
1488                hashes,
1489            };
1490            let encoded = encoded(&encodable);
1491
1492            let mut derived_buf = encoded.as_slice();
1493            let derived = decode_eth68_hashes_derived(&mut derived_buf);
1494
1495            let mut handrolled_buf = encoded.as_slice();
1496            let handrolled = NewPooledTransactionHashes68::decode(&mut handrolled_buf);
1497
1498            let handrolled_is_ok = handrolled.is_ok();
1499            prop_assert_eq!(&handrolled, &derived);
1500            if handrolled_is_ok {
1501                prop_assert!(derived_buf.is_empty());
1502                prop_assert!(handrolled_buf.is_empty());
1503            }
1504        }
1505    }
1506
1507    #[test]
1508    fn can_return_latest_block() {
1509        let mut blocks = NewBlockHashes(vec![BlockHashNumber { hash: B256::random(), number: 0 }]);
1510        let latest = blocks.latest().unwrap();
1511        assert_eq!(latest.number, 0);
1512
1513        blocks.push(BlockHashNumber { hash: B256::random(), number: 100 });
1514        blocks.push(BlockHashNumber { hash: B256::random(), number: 2 });
1515        let latest = blocks.latest().unwrap();
1516        assert_eq!(latest.number, 100);
1517    }
1518
1519    #[test]
1520    fn eth_68_tx_hash_roundtrip() {
1521        let vectors = vec![
1522            (
1523                NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] },
1524                &hex!("c380c0c0")[..],
1525            ),
1526            (
1527                NewPooledTransactionHashes68 {
1528                    types: vec![0x00],
1529                    sizes: vec![0x00],
1530                    hashes: vec![
1531                        B256::from_str(
1532                            "0x0000000000000000000000000000000000000000000000000000000000000000",
1533                        )
1534                        .unwrap(),
1535                    ],
1536                },
1537                &hex!(
1538                    "e500c180e1a00000000000000000000000000000000000000000000000000000000000000000"
1539                )[..],
1540            ),
1541            (
1542                NewPooledTransactionHashes68 {
1543                    types: vec![0x00, 0x00],
1544                    sizes: vec![0x00, 0x00],
1545                    hashes: vec![
1546                        B256::from_str(
1547                            "0x0000000000000000000000000000000000000000000000000000000000000000",
1548                        )
1549                        .unwrap(),
1550                        B256::from_str(
1551                            "0x0000000000000000000000000000000000000000000000000000000000000000",
1552                        )
1553                        .unwrap(),
1554                    ],
1555                },
1556                &hex!(
1557                    "f84a820000c28080f842a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000"
1558                )[..],
1559            ),
1560            (
1561                NewPooledTransactionHashes68 {
1562                    types: vec![0x02],
1563                    sizes: vec![0xb6],
1564                    hashes: vec![
1565                        B256::from_str(
1566                            "0xfecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124",
1567                        )
1568                        .unwrap(),
1569                    ],
1570                },
1571                &hex!(
1572                    "e602c281b6e1a0fecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124"
1573                )[..],
1574            ),
1575            (
1576                NewPooledTransactionHashes68 {
1577                    types: vec![0xff, 0xff],
1578                    sizes: vec![0xffffffff, 0xffffffff],
1579                    hashes: vec![
1580                        B256::from_str(
1581                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
1582                        )
1583                        .unwrap(),
1584                        B256::from_str(
1585                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
1586                        )
1587                        .unwrap(),
1588                    ],
1589                },
1590                &hex!(
1591                    "f85282ffffca84ffffffff84fffffffff842a0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffa0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
1592                )[..],
1593            ),
1594            (
1595                NewPooledTransactionHashes68 {
1596                    types: vec![0xff, 0xff],
1597                    sizes: vec![0xffffffff, 0xffffffff],
1598                    hashes: vec![
1599                        B256::from_str(
1600                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
1601                        )
1602                        .unwrap(),
1603                        B256::from_str(
1604                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
1605                        )
1606                        .unwrap(),
1607                    ],
1608                },
1609                &hex!(
1610                    "f85282ffffca84ffffffff84fffffffff842a0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafea0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe"
1611                )[..],
1612            ),
1613            (
1614                NewPooledTransactionHashes68 {
1615                    types: vec![0x10, 0x10],
1616                    sizes: vec![0xdeadc0de, 0xdeadc0de],
1617                    hashes: vec![
1618                        B256::from_str(
1619                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
1620                        )
1621                        .unwrap(),
1622                        B256::from_str(
1623                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
1624                        )
1625                        .unwrap(),
1626                    ],
1627                },
1628                &hex!(
1629                    "f852821010ca84deadc0de84deadc0def842a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2"
1630                )[..],
1631            ),
1632            (
1633                NewPooledTransactionHashes68 {
1634                    types: vec![0x6f, 0x6f],
1635                    sizes: vec![0x7fffffff, 0x7fffffff],
1636                    hashes: vec![
1637                        B256::from_str(
1638                            "0x0000000000000000000000000000000000000000000000000000000000000002",
1639                        )
1640                        .unwrap(),
1641                        B256::from_str(
1642                            "0x0000000000000000000000000000000000000000000000000000000000000002",
1643                        )
1644                        .unwrap(),
1645                    ],
1646                },
1647                &hex!(
1648                    "f852826f6fca847fffffff847ffffffff842a00000000000000000000000000000000000000000000000000000000000000002a00000000000000000000000000000000000000000000000000000000000000002"
1649                )[..],
1650            ),
1651        ];
1652
1653        for vector in vectors {
1654            test_encoding_vector(vector);
1655        }
1656    }
1657
1658    #[test]
1659    fn eth_72_tx_hash_roundtrip() {
1660        let vectors = vec![
1661            (
1662                NewPooledTransactionHashes72 {
1663                    types: vec![],
1664                    sizes: vec![],
1665                    hashes: vec![],
1666                    cell_mask: None,
1667                },
1668                &hex!("c480c0c080")[..],
1669            ),
1670            (
1671                NewPooledTransactionHashes72 {
1672                    types: vec![],
1673                    sizes: vec![],
1674                    hashes: vec![],
1675                    cell_mask: Some(B128::repeat_byte(0x11)),
1676                },
1677                &hex!("d480c0c09011111111111111111111111111111111")[..],
1678            ),
1679        ];
1680
1681        for vector in vectors {
1682            test_encoding_vector(vector);
1683        }
1684    }
1685
1686    #[test]
1687    fn eth_72_rejects_missing_cell_mask() {
1688        let encoded_eth68_payload = hex!("c380c0c0");
1689
1690        let result = NewPooledTransactionHashes72::decode(&mut encoded_eth68_payload.as_ref());
1691
1692        assert!(matches!(result, Err(alloy_rlp::Error::InputTooShort)));
1693    }
1694
1695    #[test]
1696    fn eth_72_dedup_preserves_message_cell_mask() {
1697        let cell_mask = Some(B128::repeat_byte(0x11));
1698        let announcement = NewPooledTransactionHashes72 {
1699            types: vec![3],
1700            sizes: vec![128],
1701            hashes: vec![B256::from([1u8; 32])],
1702            cell_mask,
1703        };
1704
1705        let partially_valid = announcement.dedup();
1706        assert_eq!(partially_valid.eth72_cell_mask(), cell_mask);
1707
1708        let valid = ValidAnnouncementData::from_partially_valid_data(partially_valid);
1709        assert_eq!(valid.eth72_cell_mask(), cell_mask);
1710    }
1711
1712    #[test]
1713    fn request_hashes_retain_count_keep_subset() {
1714        let mut hashes = RequestTxHashes::new(
1715            [
1716                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1717                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1718                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1719                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1720                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1721            ]
1722            .into_iter()
1723            .collect::<B256Set>(),
1724        );
1725
1726        let rest = hashes.retain_count(3);
1727
1728        assert_eq!(3, hashes.len());
1729        assert_eq!(2, rest.len());
1730    }
1731
1732    #[test]
1733    fn request_hashes_retain_count_keep_all() {
1734        let mut hashes = RequestTxHashes::new(
1735            [
1736                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1737                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1738                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1739                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1740                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1741            ]
1742            .into_iter()
1743            .collect::<B256Set>(),
1744        );
1745
1746        let _ = hashes.retain_count(6);
1747
1748        assert_eq!(5, hashes.len());
1749    }
1750
1751    #[test]
1752    fn split_request_hashes_keep_none() {
1753        let mut hashes = RequestTxHashes::new(
1754            [
1755                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1756                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1757                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1758                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1759                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1760            ]
1761            .into_iter()
1762            .collect::<B256Set>(),
1763        );
1764
1765        let rest = hashes.retain_count(0);
1766
1767        assert_eq!(0, hashes.len());
1768        assert_eq!(5, rest.len());
1769    }
1770
1771    fn signed_transaction() -> impl SignedTransaction {
1772        TransactionSigned::new_unhashed(
1773            Transaction::Legacy(Default::default()),
1774            Signature::new(
1775                U256::from_str(
1776                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c12",
1777                )
1778                .unwrap(),
1779                U256::from_str(
1780                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c10",
1781                )
1782                .unwrap(),
1783                false,
1784            ),
1785        )
1786    }
1787
1788    #[test]
1789    fn test_pooled_tx_hashes_68_push() {
1790        let tx = signed_transaction();
1791        let mut tx_hashes =
1792            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1793        tx_hashes.push(&tx);
1794        assert_eq!(tx_hashes.types.len(), 1);
1795        assert_eq!(tx_hashes.sizes.len(), 1);
1796        assert_eq!(tx_hashes.hashes.len(), 1);
1797        assert_eq!(tx_hashes.types[0], tx.ty());
1798        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1799        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1800    }
1801
1802    #[test]
1803    fn test_pooled_tx_hashes_68_extend() {
1804        let tx = signed_transaction();
1805        let txs = vec![tx.clone(), tx.clone()];
1806        let mut tx_hashes =
1807            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1808        tx_hashes.extend(&txs);
1809        assert_eq!(tx_hashes.types.len(), 2);
1810        assert_eq!(tx_hashes.sizes.len(), 2);
1811        assert_eq!(tx_hashes.hashes.len(), 2);
1812        assert_eq!(tx_hashes.types[0], tx.ty());
1813        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1814        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1815        assert_eq!(tx_hashes.types[1], tx.ty());
1816        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1817        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1818    }
1819
1820    #[test]
1821    fn test_pooled_tx_hashes_68_with_transaction() {
1822        let tx = signed_transaction();
1823        let tx_hashes =
1824            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1825                .with_transaction(&tx);
1826        assert_eq!(tx_hashes.types.len(), 1);
1827        assert_eq!(tx_hashes.sizes.len(), 1);
1828        assert_eq!(tx_hashes.hashes.len(), 1);
1829        assert_eq!(tx_hashes.types[0], tx.ty());
1830        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1831        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1832    }
1833
1834    #[test]
1835    fn test_pooled_tx_hashes_68_with_transactions() {
1836        let tx = signed_transaction();
1837        let txs = vec![tx.clone(), tx.clone()];
1838        let tx_hashes =
1839            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1840                .with_transactions(&txs);
1841        assert_eq!(tx_hashes.types.len(), 2);
1842        assert_eq!(tx_hashes.sizes.len(), 2);
1843        assert_eq!(tx_hashes.hashes.len(), 2);
1844        assert_eq!(tx_hashes.types[0], tx.ty());
1845        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1846        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1847        assert_eq!(tx_hashes.types[1], tx.ty());
1848        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1849        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1850    }
1851}
reth_eth_wire_types/broadcast.rs

reth_eth_wire_types/
broadcast.rs
