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, .. } = 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(deduped_data)
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}
1165
1166handle_mempool_data_map_impl!(PartiallyValidData<V>, <V>);
1167
1168impl<V> PartiallyValidData<V> {
1169    /// Wraps raw data.
1170    pub const fn from_raw_data(data: B256Map<V>, version: Option<EthVersion>) -> Self {
1171        Self { data, version }
1172    }
1173
1174    /// Wraps raw data with version [`EthVersion::Eth72`].
1175    pub const fn from_raw_data_eth72(data: B256Map<V>) -> Self {
1176        Self::from_raw_data(data, Some(EthVersion::Eth72))
1177    }
1178
1179    /// Wraps raw data with version [`EthVersion::Eth68`].
1180    pub const fn from_raw_data_eth68(data: B256Map<V>) -> Self {
1181        Self::from_raw_data(data, Some(EthVersion::Eth68))
1182    }
1183
1184    /// Wraps raw data with version [`EthVersion::Eth66`].
1185    pub const fn from_raw_data_eth66(data: B256Map<V>) -> Self {
1186        Self::from_raw_data(data, Some(EthVersion::Eth66))
1187    }
1188
1189    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth72`](EthVersion::Eth72)
1190    /// announcement.
1191    pub fn empty_eth72() -> Self {
1192        Self::from_raw_data_eth72(B256Map::default())
1193    }
1194
1195    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth68`](EthVersion::Eth68)
1196    /// announcement.
1197    pub fn empty_eth68() -> Self {
1198        Self::from_raw_data_eth68(B256Map::default())
1199    }
1200
1201    /// Returns a new [`PartiallyValidData`] with empty data from an [`Eth66`](EthVersion::Eth66)
1202    /// announcement.
1203    pub fn empty_eth66() -> Self {
1204        Self::from_raw_data_eth66(B256Map::default())
1205    }
1206
1207    /// Returns the version of the message this data was received in if different versions of the
1208    /// message exist.
1209    pub const fn msg_version(&self) -> Option<EthVersion> {
1210        self.version
1211    }
1212
1213    /// Destructs returning the validated data.
1214    pub fn into_data(self) -> B256Map<V> {
1215        self.data
1216    }
1217}
1218
1219/// Partially validated data from an announcement or a
1220/// [`PooledTransactions`](crate::PooledTransactions) response.
1221#[derive(Debug, Deref, DerefMut, IntoIterator, From)]
1222pub struct ValidAnnouncementData {
1223    #[deref]
1224    #[deref_mut]
1225    #[into_iterator]
1226    data: B256Map<Eth68TxMetadata>,
1227    version: EthVersion,
1228}
1229
1230handle_mempool_data_map_impl!(ValidAnnouncementData,);
1231
1232impl ValidAnnouncementData {
1233    /// Destructs returning only the valid hashes and the announcement message version. Caution! If
1234    /// this is [`Eth68`](EthVersion::Eth68) announcement data, this drops the metadata.
1235    pub fn into_request_hashes(self) -> (RequestTxHashes, EthVersion) {
1236        let hashes = self.data.into_keys().collect::<B256Set>();
1237
1238        (RequestTxHashes::new(hashes), self.version)
1239    }
1240
1241    /// Conversion from [`PartiallyValidData`] from an announcement. Note! [`PartiallyValidData`]
1242    /// from an announcement, should have some [`EthVersion`]. Panics if [`PartiallyValidData`] has
1243    /// version set to `None`.
1244    pub fn from_partially_valid_data(data: PartiallyValidData<Eth68TxMetadata>) -> Self {
1245        let PartiallyValidData { data, version } = data;
1246
1247        let version = version.expect("should have eth version for conversion");
1248
1249        Self { data, version }
1250    }
1251
1252    /// Destructs returning the validated data.
1253    pub fn into_data(self) -> B256Map<Eth68TxMetadata> {
1254        self.data
1255    }
1256}
1257
1258impl HandleVersionedMempoolData for ValidAnnouncementData {
1259    fn msg_version(&self) -> EthVersion {
1260        self.version
1261    }
1262}
1263
1264/// Hashes to request from a peer.
1265#[derive(Debug, Default, Deref, DerefMut, IntoIterator, Constructor)]
1266pub struct RequestTxHashes {
1267    #[deref]
1268    #[deref_mut]
1269    #[into_iterator(owned, ref)]
1270    hashes: B256Set,
1271}
1272
1273impl RequestTxHashes {
1274    /// Returns a new [`RequestTxHashes`] with given capacity for hashes. Caution! Make sure to
1275    /// call `shrink_to_fit` on [`RequestTxHashes`] when full, especially where it will
1276    /// be stored in its entirety like in the future waiting for a
1277    /// [`GetPooledTransactions`](crate::GetPooledTransactions) request to resolve.
1278    pub fn with_capacity(capacity: usize) -> Self {
1279        Self::new(B256Set::with_capacity_and_hasher(capacity, Default::default()))
1280    }
1281
1282    /// Returns a new empty instance.
1283    fn empty() -> Self {
1284        Self::new(B256Set::default())
1285    }
1286
1287    /// Retains the given number of elements, returning an iterator over the rest.
1288    pub fn retain_count(&mut self, count: usize) -> Self {
1289        let rest_capacity = self.hashes.len().saturating_sub(count);
1290        if rest_capacity == 0 {
1291            return Self::empty()
1292        }
1293        let mut rest = Self::with_capacity(rest_capacity);
1294
1295        let mut i = 0;
1296        self.hashes.retain(|hash| {
1297            if i >= count {
1298                rest.insert(*hash);
1299                return false
1300            }
1301            i += 1;
1302
1303            true
1304        });
1305
1306        rest
1307    }
1308}
1309
1310impl FromIterator<(TxHash, Eth68TxMetadata)> for RequestTxHashes {
1311    fn from_iter<I: IntoIterator<Item = (TxHash, Eth68TxMetadata)>>(iter: I) -> Self {
1312        Self::new(iter.into_iter().map(|(hash, _)| hash).collect())
1313    }
1314}
1315
1316/// The earliest block, the latest block and hash of the latest block which can be provided.
1317/// See [BlockRangeUpdate](https://github.com/ethereum/devp2p/blob/master/caps/eth.md#blockrangeupdate-0x11).
1318#[derive(Clone, Debug, PartialEq, Eq, Default, RlpEncodable, RlpDecodable)]
1319#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
1320#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))]
1321pub struct BlockRangeUpdate {
1322    /// The earliest block which is available.
1323    pub earliest: u64,
1324    /// The latest block which is available.
1325    pub latest: u64,
1326    /// Latest available block's hash.
1327    pub latest_hash: B256,
1328}
1329
1330impl InMemorySize for NewPooledTransactionHashes {
1331    fn size(&self) -> usize {
1332        match self {
1333            Self::Eth66(msg) => msg.0.len() * core::mem::size_of::<B256>(),
1334            Self::Eth68(msg) => {
1335                msg.types.len() * core::mem::size_of::<u8>() +
1336                    msg.sizes.len() * core::mem::size_of::<usize>() +
1337                    msg.hashes.len() * core::mem::size_of::<B256>()
1338            }
1339            Self::Eth72(msg) => {
1340                msg.types.len() * core::mem::size_of::<u8>() +
1341                    msg.sizes.len() * core::mem::size_of::<usize>() +
1342                    msg.hashes.len() * core::mem::size_of::<B256>() +
1343                    core::mem::size_of::<B128>()
1344            }
1345        }
1346    }
1347}
1348
1349#[cfg(test)]
1350mod tests {
1351    use super::*;
1352    use alloy_consensus::{transaction::TxHashRef, Typed2718};
1353    use alloy_eips::eip2718::Encodable2718;
1354    use alloy_primitives::{b256, hex, Bytes, Signature, U256};
1355    use alloy_rlp::{RlpDecodable, RlpEncodable};
1356    use proptest::prelude::*;
1357    use reth_ethereum_primitives::{Transaction, TransactionSigned};
1358    use std::str::FromStr;
1359
1360    /// Takes as input a struct / encoded hex message pair, ensuring that we encode to the exact hex
1361    /// message, and decode to the exact struct.
1362    fn test_encoding_vector<T: Encodable + Decodable + PartialEq + core::fmt::Debug>(
1363        input: (T, &[u8]),
1364    ) {
1365        let (expected_decoded, expected_encoded) = input;
1366        let mut encoded = Vec::new();
1367        expected_decoded.encode(&mut encoded);
1368
1369        assert_eq!(hex::encode(&encoded), hex::encode(expected_encoded));
1370
1371        let decoded = T::decode(&mut encoded.as_ref()).unwrap();
1372        assert_eq!(expected_decoded, decoded);
1373    }
1374
1375    fn encoded<T: Encodable>(value: &T) -> Vec<u8> {
1376        let mut out = Vec::new();
1377        value.encode(&mut out);
1378        out
1379    }
1380
1381    #[derive(RlpEncodable, RlpDecodable)]
1382    struct EncodableNewPooledTransactionHashes68 {
1383        types: Bytes,
1384        sizes: Vec<usize>,
1385        hashes: Vec<B256>,
1386    }
1387
1388    type NewPooledTransactionHashes68Fields = (Vec<u8>, Vec<usize>, Vec<B256>);
1389
1390    fn decode_eth68_hashes_derived(
1391        buf: &mut &[u8],
1392    ) -> alloy_rlp::Result<NewPooledTransactionHashes68> {
1393        let encodable = EncodableNewPooledTransactionHashes68::decode(buf)?;
1394        let msg = NewPooledTransactionHashes68 {
1395            types: encodable.types.into(),
1396            sizes: encodable.sizes,
1397            hashes: encodable.hashes,
1398        };
1399
1400        ensure_pooled_transaction_hashes_lengths(
1401            msg.hashes.len(),
1402            msg.types.len(),
1403            msg.sizes.len(),
1404        )?;
1405
1406        Ok(msg)
1407    }
1408
1409    fn eth68_hash_fields_strategy() -> impl Strategy<Value = NewPooledTransactionHashes68Fields> {
1410        (0usize..128, 0usize..128, 0usize..128).prop_flat_map(
1411            |(types_len, sizes_len, hashes_len)| {
1412                (
1413                    proptest::collection::vec(any::<u8>(), types_len),
1414                    proptest::collection::vec(0usize..131_072, sizes_len),
1415                    proptest::collection::vec(any::<B256>(), hashes_len),
1416                )
1417            },
1418        )
1419    }
1420
1421    proptest! {
1422        #[test]
1423        fn broadcast_pool_transactions_match_shared_transactions_encoding(
1424            txs in proptest::collection::vec(
1425                proptest_arbitrary_interop::arb::<TransactionSigned>(),
1426                0..32,
1427            )
1428        ) {
1429            let shared = SharedTransactions::<TransactionSigned>(
1430                txs.iter().cloned().map(Arc::new).collect(),
1431            );
1432            let broadcast = BroadcastPoolTransactions(
1433                txs.iter().cloned().map(LazyEncoded::new).collect(),
1434            );
1435
1436            prop_assert_eq!(broadcast.length(), shared.length());
1437
1438            let shared_encoded = encoded(&shared);
1439            let broadcast_encoded = encoded(&broadcast);
1440            prop_assert_eq!(&broadcast_encoded, &shared_encoded);
1441
1442            let broadcast_encoded_cached = encoded(&broadcast);
1443            prop_assert_eq!(&broadcast_encoded_cached, &shared_encoded);
1444
1445            let mut shared_bytes = shared_encoded.as_slice();
1446            let decoded_shared = SharedTransactions::<TransactionSigned>::decode(&mut shared_bytes)
1447                .expect("shared transactions decode");
1448            prop_assert!(shared_bytes.is_empty());
1449
1450            let mut broadcast_bytes = broadcast_encoded.as_slice();
1451            let decoded_broadcast =
1452                SharedTransactions::<TransactionSigned>::decode(&mut broadcast_bytes)
1453                    .expect("broadcast pool transactions decode as shared transactions");
1454            prop_assert!(broadcast_bytes.is_empty());
1455
1456            prop_assert_eq!(decoded_broadcast, decoded_shared);
1457        }
1458
1459        #[test]
1460        fn eth_68_handrolled_decode_matches_derived_implementation(
1461            (types, sizes, hashes) in eth68_hash_fields_strategy()
1462        ) {
1463            let encodable = EncodableNewPooledTransactionHashes68 {
1464                types: Bytes::from(types),
1465                sizes,
1466                hashes,
1467            };
1468            let encoded = encoded(&encodable);
1469
1470            let mut derived_buf = encoded.as_slice();
1471            let derived = decode_eth68_hashes_derived(&mut derived_buf);
1472
1473            let mut handrolled_buf = encoded.as_slice();
1474            let handrolled = NewPooledTransactionHashes68::decode(&mut handrolled_buf);
1475
1476            let handrolled_is_ok = handrolled.is_ok();
1477            prop_assert_eq!(&handrolled, &derived);
1478            if handrolled_is_ok {
1479                prop_assert!(derived_buf.is_empty());
1480                prop_assert!(handrolled_buf.is_empty());
1481            }
1482        }
1483    }
1484
1485    #[test]
1486    fn can_return_latest_block() {
1487        let mut blocks = NewBlockHashes(vec![BlockHashNumber { hash: B256::random(), number: 0 }]);
1488        let latest = blocks.latest().unwrap();
1489        assert_eq!(latest.number, 0);
1490
1491        blocks.push(BlockHashNumber { hash: B256::random(), number: 100 });
1492        blocks.push(BlockHashNumber { hash: B256::random(), number: 2 });
1493        let latest = blocks.latest().unwrap();
1494        assert_eq!(latest.number, 100);
1495    }
1496
1497    #[test]
1498    fn eth_68_tx_hash_roundtrip() {
1499        let vectors = vec![
1500            (
1501                NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] },
1502                &hex!("c380c0c0")[..],
1503            ),
1504            (
1505                NewPooledTransactionHashes68 {
1506                    types: vec![0x00],
1507                    sizes: vec![0x00],
1508                    hashes: vec![
1509                        B256::from_str(
1510                            "0x0000000000000000000000000000000000000000000000000000000000000000",
1511                        )
1512                        .unwrap(),
1513                    ],
1514                },
1515                &hex!(
1516                    "e500c180e1a00000000000000000000000000000000000000000000000000000000000000000"
1517                )[..],
1518            ),
1519            (
1520                NewPooledTransactionHashes68 {
1521                    types: vec![0x00, 0x00],
1522                    sizes: vec![0x00, 0x00],
1523                    hashes: vec![
1524                        B256::from_str(
1525                            "0x0000000000000000000000000000000000000000000000000000000000000000",
1526                        )
1527                        .unwrap(),
1528                        B256::from_str(
1529                            "0x0000000000000000000000000000000000000000000000000000000000000000",
1530                        )
1531                        .unwrap(),
1532                    ],
1533                },
1534                &hex!(
1535                    "f84a820000c28080f842a00000000000000000000000000000000000000000000000000000000000000000a00000000000000000000000000000000000000000000000000000000000000000"
1536                )[..],
1537            ),
1538            (
1539                NewPooledTransactionHashes68 {
1540                    types: vec![0x02],
1541                    sizes: vec![0xb6],
1542                    hashes: vec![
1543                        B256::from_str(
1544                            "0xfecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124",
1545                        )
1546                        .unwrap(),
1547                    ],
1548                },
1549                &hex!(
1550                    "e602c281b6e1a0fecbed04c7b88d8e7221a0a3f5dc33f220212347fc167459ea5cc9c3eb4c1124"
1551                )[..],
1552            ),
1553            (
1554                NewPooledTransactionHashes68 {
1555                    types: vec![0xff, 0xff],
1556                    sizes: vec![0xffffffff, 0xffffffff],
1557                    hashes: vec![
1558                        B256::from_str(
1559                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
1560                        )
1561                        .unwrap(),
1562                        B256::from_str(
1563                            "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
1564                        )
1565                        .unwrap(),
1566                    ],
1567                },
1568                &hex!(
1569                    "f85282ffffca84ffffffff84fffffffff842a0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffa0ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
1570                )[..],
1571            ),
1572            (
1573                NewPooledTransactionHashes68 {
1574                    types: vec![0xff, 0xff],
1575                    sizes: vec![0xffffffff, 0xffffffff],
1576                    hashes: vec![
1577                        B256::from_str(
1578                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
1579                        )
1580                        .unwrap(),
1581                        B256::from_str(
1582                            "0xbeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe",
1583                        )
1584                        .unwrap(),
1585                    ],
1586                },
1587                &hex!(
1588                    "f85282ffffca84ffffffff84fffffffff842a0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafea0beefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafebeefcafe"
1589                )[..],
1590            ),
1591            (
1592                NewPooledTransactionHashes68 {
1593                    types: vec![0x10, 0x10],
1594                    sizes: vec![0xdeadc0de, 0xdeadc0de],
1595                    hashes: vec![
1596                        B256::from_str(
1597                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
1598                        )
1599                        .unwrap(),
1600                        B256::from_str(
1601                            "0x3b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2",
1602                        )
1603                        .unwrap(),
1604                    ],
1605                },
1606                &hex!(
1607                    "f852821010ca84deadc0de84deadc0def842a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2a03b9aca00f0671c9a2a1b817a0a78d3fe0c0f776cccb2a8c3c1b412a4f4e4d4e2"
1608                )[..],
1609            ),
1610            (
1611                NewPooledTransactionHashes68 {
1612                    types: vec![0x6f, 0x6f],
1613                    sizes: vec![0x7fffffff, 0x7fffffff],
1614                    hashes: vec![
1615                        B256::from_str(
1616                            "0x0000000000000000000000000000000000000000000000000000000000000002",
1617                        )
1618                        .unwrap(),
1619                        B256::from_str(
1620                            "0x0000000000000000000000000000000000000000000000000000000000000002",
1621                        )
1622                        .unwrap(),
1623                    ],
1624                },
1625                &hex!(
1626                    "f852826f6fca847fffffff847ffffffff842a00000000000000000000000000000000000000000000000000000000000000002a00000000000000000000000000000000000000000000000000000000000000002"
1627                )[..],
1628            ),
1629        ];
1630
1631        for vector in vectors {
1632            test_encoding_vector(vector);
1633        }
1634    }
1635
1636    #[test]
1637    fn eth_72_tx_hash_roundtrip() {
1638        let vectors = vec![
1639            (
1640                NewPooledTransactionHashes72 {
1641                    types: vec![],
1642                    sizes: vec![],
1643                    hashes: vec![],
1644                    cell_mask: None,
1645                },
1646                &hex!("c480c0c080")[..],
1647            ),
1648            (
1649                NewPooledTransactionHashes72 {
1650                    types: vec![],
1651                    sizes: vec![],
1652                    hashes: vec![],
1653                    cell_mask: Some(B128::repeat_byte(0x11)),
1654                },
1655                &hex!("d480c0c09011111111111111111111111111111111")[..],
1656            ),
1657        ];
1658
1659        for vector in vectors {
1660            test_encoding_vector(vector);
1661        }
1662    }
1663
1664    #[test]
1665    fn eth_72_rejects_missing_cell_mask() {
1666        let encoded_eth68_payload = hex!("c380c0c0");
1667
1668        let result = NewPooledTransactionHashes72::decode(&mut encoded_eth68_payload.as_ref());
1669
1670        assert!(matches!(result, Err(alloy_rlp::Error::InputTooShort)));
1671    }
1672
1673    #[test]
1674    fn request_hashes_retain_count_keep_subset() {
1675        let mut hashes = RequestTxHashes::new(
1676            [
1677                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1678                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1679                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1680                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1681                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1682            ]
1683            .into_iter()
1684            .collect::<B256Set>(),
1685        );
1686
1687        let rest = hashes.retain_count(3);
1688
1689        assert_eq!(3, hashes.len());
1690        assert_eq!(2, rest.len());
1691    }
1692
1693    #[test]
1694    fn request_hashes_retain_count_keep_all() {
1695        let mut hashes = RequestTxHashes::new(
1696            [
1697                b256!("0x0000000000000000000000000000000000000000000000000000000000000001"),
1698                b256!("0x0000000000000000000000000000000000000000000000000000000000000002"),
1699                b256!("0x0000000000000000000000000000000000000000000000000000000000000003"),
1700                b256!("0x0000000000000000000000000000000000000000000000000000000000000004"),
1701                b256!("0x0000000000000000000000000000000000000000000000000000000000000005"),
1702            ]
1703            .into_iter()
1704            .collect::<B256Set>(),
1705        );
1706
1707        let _ = hashes.retain_count(6);
1708
1709        assert_eq!(5, hashes.len());
1710    }
1711
1712    #[test]
1713    fn split_request_hashes_keep_none() {
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(0);
1727
1728        assert_eq!(0, hashes.len());
1729        assert_eq!(5, rest.len());
1730    }
1731
1732    fn signed_transaction() -> impl SignedTransaction {
1733        TransactionSigned::new_unhashed(
1734            Transaction::Legacy(Default::default()),
1735            Signature::new(
1736                U256::from_str(
1737                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c12",
1738                )
1739                .unwrap(),
1740                U256::from_str(
1741                    "0x64b1702d9298fee62dfeccc57d322a463ad55ca201256d01f62b45b2e1c21c10",
1742                )
1743                .unwrap(),
1744                false,
1745            ),
1746        )
1747    }
1748
1749    #[test]
1750    fn test_pooled_tx_hashes_68_push() {
1751        let tx = signed_transaction();
1752        let mut tx_hashes =
1753            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1754        tx_hashes.push(&tx);
1755        assert_eq!(tx_hashes.types.len(), 1);
1756        assert_eq!(tx_hashes.sizes.len(), 1);
1757        assert_eq!(tx_hashes.hashes.len(), 1);
1758        assert_eq!(tx_hashes.types[0], tx.ty());
1759        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1760        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1761    }
1762
1763    #[test]
1764    fn test_pooled_tx_hashes_68_extend() {
1765        let tx = signed_transaction();
1766        let txs = vec![tx.clone(), tx.clone()];
1767        let mut tx_hashes =
1768            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] };
1769        tx_hashes.extend(&txs);
1770        assert_eq!(tx_hashes.types.len(), 2);
1771        assert_eq!(tx_hashes.sizes.len(), 2);
1772        assert_eq!(tx_hashes.hashes.len(), 2);
1773        assert_eq!(tx_hashes.types[0], tx.ty());
1774        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1775        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1776        assert_eq!(tx_hashes.types[1], tx.ty());
1777        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1778        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1779    }
1780
1781    #[test]
1782    fn test_pooled_tx_hashes_68_with_transaction() {
1783        let tx = signed_transaction();
1784        let tx_hashes =
1785            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1786                .with_transaction(&tx);
1787        assert_eq!(tx_hashes.types.len(), 1);
1788        assert_eq!(tx_hashes.sizes.len(), 1);
1789        assert_eq!(tx_hashes.hashes.len(), 1);
1790        assert_eq!(tx_hashes.types[0], tx.ty());
1791        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1792        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1793    }
1794
1795    #[test]
1796    fn test_pooled_tx_hashes_68_with_transactions() {
1797        let tx = signed_transaction();
1798        let txs = vec![tx.clone(), tx.clone()];
1799        let tx_hashes =
1800            NewPooledTransactionHashes68 { types: vec![], sizes: vec![], hashes: vec![] }
1801                .with_transactions(&txs);
1802        assert_eq!(tx_hashes.types.len(), 2);
1803        assert_eq!(tx_hashes.sizes.len(), 2);
1804        assert_eq!(tx_hashes.hashes.len(), 2);
1805        assert_eq!(tx_hashes.types[0], tx.ty());
1806        assert_eq!(tx_hashes.sizes[0], tx.encode_2718_len());
1807        assert_eq!(tx_hashes.hashes[0], *tx.tx_hash());
1808        assert_eq!(tx_hashes.types[1], tx.ty());
1809        assert_eq!(tx_hashes.sizes[1], tx.encode_2718_len());
1810        assert_eq!(tx_hashes.hashes[1], *tx.tx_hash());
1811    }
1812}
reth_eth_wire_types/broadcast.rs

reth_eth_wire_types/
broadcast.rs
