reth_eth_wire_types/
message.rs

1//! Implements Ethereum wire protocol for versions 66 through 71.
2//! Defines structs/enums for messages, request-response pairs, and broadcasts.
3//! Handles compatibility with [`EthVersion`].
4//!
5//! Examples include creating, encoding, and decoding protocol messages.
6//!
7//! Reference: [Ethereum Wire Protocol](https://github.com/ethereum/devp2p/blob/master/caps/eth.md).
8
9use super::{
10    broadcast::NewBlockHashes, BlockAccessLists, BlockBodies, BlockHeaders, GetBlockAccessLists,
11    GetBlockBodies, GetBlockHeaders, GetNodeData, GetPooledTransactions, GetReceipts,
12    GetReceipts70, NewPooledTransactionHashes66, NewPooledTransactionHashes68, NodeData,
13    PooledTransactions, Receipts, Status, StatusEth69, Transactions,
14};
15use crate::{
16    status::StatusMessage, BlockRangeUpdate, BroadcastPoolTransactions, Cells,
17    EthNetworkPrimitives, EthVersion, GetCells, NetworkPrimitives, NewPooledTransactionHashes72,
18    RawCapabilityMessage, Receipts69, Receipts70, SharedTransactions,
19};
20use alloc::{boxed::Box, string::String, sync::Arc};
21use alloy_primitives::{
22    bytes::{Buf, BufMut},
23    Bytes,
24};
25use alloy_rlp::{Decodable, Encodable, Header};
26use core::fmt::Debug;
27
28/// [`MAX_MESSAGE_SIZE`] is the maximum cap on the size of a protocol message.
29// https://github.com/ethereum/go-ethereum/blob/30602163d5d8321fbc68afdcbbaf2362b2641bde/eth/protocols/eth/protocol.go#L50
30pub const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024;
31
32/// Multiplier applied to `max_message_size` to derive the in-memory budget for decoding
33/// `Transactions` and `PooledTransactions` messages.
34///
35/// Decoded transactions expand relative to their RLP encoding due to struct overhead and heap
36/// allocations. With many peers in flight this can cause significant memory pressure, so we
37/// stop decoding once the cumulative in-memory size of decoded transactions exceeds
38/// `max_message_size * TX_MEMORY_BUDGET_MULTIPLIER`. Remaining transactions are silently dropped.
39pub const TX_MEMORY_BUDGET_MULTIPLIER: usize = 2;
40
41/// Error when sending/receiving a message
42#[derive(thiserror::Error, Debug)]
43pub enum MessageError {
44    /// Flags an unrecognized message ID for a given protocol version.
45    #[error("message id {1:?} is invalid for version {0:?}")]
46    Invalid(EthVersion, EthMessageID),
47    /// Expected a Status message but received a different message type.
48    #[error("expected status message but received {0:?}")]
49    ExpectedStatusMessage(EthMessageID),
50    /// Thrown when rlp decoding a message failed.
51    #[error("RLP error: {0}")]
52    RlpError(#[from] alloy_rlp::Error),
53    /// Other message error with custom message
54    #[error("{0}")]
55    Other(String),
56}
57
58/// An `eth` protocol message, containing a message ID and payload.
59#[derive(Clone, Debug, PartialEq, Eq)]
60#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
61pub struct ProtocolMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
62    /// The unique identifier representing the type of the Ethereum message.
63    pub message_type: EthMessageID,
64    /// The content of the message, including specific data based on the message type.
65    #[cfg_attr(
66        feature = "serde",
67        serde(bound = "EthMessage<N>: serde::Serialize + serde::de::DeserializeOwned")
68    )]
69    pub message: EthMessage<N>,
70}
71
72impl<N: NetworkPrimitives> ProtocolMessage<N> {
73    /// Decode only a Status message from RLP bytes.
74    ///
75    /// This is used during the eth handshake where only a Status message is a valid response.
76    /// Returns an error if the message is not a Status message.
77    pub fn decode_status(
78        version: EthVersion,
79        buf: &mut &[u8],
80    ) -> Result<StatusMessage, MessageError> {
81        let message_type = EthMessageID::decode(buf)?;
82
83        if message_type != EthMessageID::Status {
84            return Err(MessageError::ExpectedStatusMessage(message_type))
85        }
86
87        let status = if version < EthVersion::Eth69 {
88            StatusMessage::Legacy(Status::decode(buf)?)
89        } else {
90            StatusMessage::Eth69(StatusEth69::decode(buf)?)
91        };
92
93        Ok(status)
94    }
95
96    /// Create a new `ProtocolMessage` from a message type and message rlp bytes.
97    ///
98    /// This will enforce decoding according to the given [`EthVersion`] of the connection.
99    pub fn decode_message(version: EthVersion, buf: &mut &[u8]) -> Result<Self, MessageError> {
100        Self::decode_message_with_tx_memory_budget(version, buf, usize::MAX)
101    }
102
103    /// Like [`Self::decode_message`], but caps the cumulative in-memory size of decoded
104    /// transactions in `Transactions` and `PooledTransactions` messages. Once exceeded,
105    /// remaining transactions are silently dropped.
106    ///
107    /// Use [`TX_MEMORY_BUDGET_MULTIPLIER`] to derive a reasonable default.
108    pub fn decode_message_with_tx_memory_budget(
109        version: EthVersion,
110        buf: &mut &[u8],
111        tx_memory_budget: usize,
112    ) -> Result<Self, MessageError> {
113        let message_type = EthMessageID::decode(buf)?;
114
115        // For EIP-7642 (https://github.com/ethereum/EIPs/blob/master/EIPS/eip-7642.md):
116        // pre-merge (legacy) status messages include total difficulty, whereas eth/69 omits it.
117        let message = match message_type {
118            EthMessageID::Status => EthMessage::Status(if version < EthVersion::Eth69 {
119                StatusMessage::Legacy(Status::decode(buf)?)
120            } else {
121                StatusMessage::Eth69(StatusEth69::decode(buf)?)
122            }),
123            EthMessageID::NewBlockHashes => {
124                EthMessage::NewBlockHashes(NewBlockHashes::decode(buf)?)
125            }
126            EthMessageID::NewBlock => {
127                EthMessage::NewBlock(Box::new(N::NewBlockPayload::decode(buf)?))
128            }
129            EthMessageID::Transactions => EthMessage::Transactions(
130                Transactions::decode_with_memory_budget(buf, tx_memory_budget)?,
131            ),
132            EthMessageID::NewPooledTransactionHashes => {
133                if version >= EthVersion::Eth72 {
134                    EthMessage::NewPooledTransactionHashes72(NewPooledTransactionHashes72::decode(
135                        buf,
136                    )?)
137                } else if version >= EthVersion::Eth68 {
138                    EthMessage::NewPooledTransactionHashes68(NewPooledTransactionHashes68::decode(
139                        buf,
140                    )?)
141                } else {
142                    EthMessage::NewPooledTransactionHashes66(NewPooledTransactionHashes66::decode(
143                        buf,
144                    )?)
145                }
146            }
147            EthMessageID::GetBlockHeaders => EthMessage::GetBlockHeaders(RequestPair::decode(buf)?),
148            EthMessageID::BlockHeaders => EthMessage::BlockHeaders(RequestPair::decode(buf)?),
149            EthMessageID::GetBlockBodies => EthMessage::GetBlockBodies(RequestPair::decode(buf)?),
150            EthMessageID::BlockBodies => EthMessage::BlockBodies(RequestPair::decode(buf)?),
151            EthMessageID::GetPooledTransactions => {
152                EthMessage::GetPooledTransactions(RequestPair::decode(buf)?)
153            }
154            EthMessageID::PooledTransactions => {
155                EthMessage::PooledTransactions(RequestPair::decode_with(buf, |buf| {
156                    PooledTransactions::decode_with_memory_budget(buf, tx_memory_budget)
157                })?)
158            }
159            EthMessageID::GetNodeData => {
160                if version >= EthVersion::Eth67 {
161                    return Err(MessageError::Invalid(version, EthMessageID::GetNodeData))
162                }
163                EthMessage::GetNodeData(RequestPair::decode(buf)?)
164            }
165            EthMessageID::NodeData => {
166                if version >= EthVersion::Eth67 {
167                    return Err(MessageError::Invalid(version, EthMessageID::NodeData))
168                }
169                EthMessage::NodeData(RequestPair::decode(buf)?)
170            }
171            EthMessageID::GetReceipts => {
172                if version >= EthVersion::Eth70 {
173                    EthMessage::GetReceipts70(RequestPair::decode(buf)?)
174                } else {
175                    EthMessage::GetReceipts(RequestPair::decode(buf)?)
176                }
177            }
178            EthMessageID::Receipts => {
179                match version {
180                    v if v >= EthVersion::Eth70 => {
181                        // eth/70 continues to omit bloom filters and adds the
182                        // `lastBlockIncomplete` flag, encoded as
183                        // `[request-id, lastBlockIncomplete, [[receipt₁, receipt₂], ...]]`.
184                        EthMessage::Receipts70(RequestPair::decode(buf)?)
185                    }
186                    EthVersion::Eth69 => {
187                        // with eth69, receipts no longer include the bloom
188                        EthMessage::Receipts69(RequestPair::decode(buf)?)
189                    }
190                    _ => {
191                        // before eth69 we need to decode the bloom  as well
192                        EthMessage::Receipts(RequestPair::decode(buf)?)
193                    }
194                }
195            }
196            EthMessageID::BlockRangeUpdate => {
197                if version < EthVersion::Eth69 {
198                    return Err(MessageError::Invalid(version, EthMessageID::BlockRangeUpdate))
199                }
200                EthMessage::BlockRangeUpdate(BlockRangeUpdate::decode(buf)?)
201            }
202            EthMessageID::GetBlockAccessLists => {
203                if version < EthVersion::Eth71 {
204                    return Err(MessageError::Invalid(version, EthMessageID::GetBlockAccessLists))
205                }
206                EthMessage::GetBlockAccessLists(RequestPair::decode(buf)?)
207            }
208            EthMessageID::BlockAccessLists => {
209                if version < EthVersion::Eth71 {
210                    return Err(MessageError::Invalid(version, EthMessageID::BlockAccessLists))
211                }
212                EthMessage::BlockAccessLists(RequestPair::decode(buf)?)
213            }
214            EthMessageID::Cells => {
215                if version < EthVersion::Eth72 {
216                    return Err(MessageError::Invalid(version, EthMessageID::Cells))
217                }
218                EthMessage::Cells(RequestPair::decode(buf)?)
219            }
220            EthMessageID::GetCells => {
221                if version < EthVersion::Eth72 {
222                    return Err(MessageError::Invalid(version, EthMessageID::GetCells))
223                }
224                EthMessage::GetCells(RequestPair::decode(buf)?)
225            }
226            EthMessageID::Other(_) => {
227                let raw_payload = Bytes::copy_from_slice(buf);
228                buf.advance(raw_payload.len());
229                EthMessage::Other(RawCapabilityMessage::new(
230                    message_type.to_u8() as usize,
231                    raw_payload.into(),
232                ))
233            }
234        };
235        Ok(Self { message_type, message })
236    }
237}
238
239impl<N: NetworkPrimitives> Encodable for ProtocolMessage<N> {
240    /// Encodes the protocol message into bytes. The message type is encoded as a single byte and
241    /// prepended to the message.
242    fn encode(&self, out: &mut dyn BufMut) {
243        self.message_type.encode(out);
244        self.message.encode(out);
245    }
246    fn length(&self) -> usize {
247        self.message_type.length() + self.message.length()
248    }
249}
250
251impl<N: NetworkPrimitives> From<EthMessage<N>> for ProtocolMessage<N> {
252    fn from(message: EthMessage<N>) -> Self {
253        Self { message_type: message.message_id(), message }
254    }
255}
256
257/// Represents messages that can be sent to multiple peers.
258#[derive(Clone, Debug)]
259pub struct ProtocolBroadcastMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
260    /// The unique identifier representing the type of the Ethereum message.
261    pub message_type: EthMessageID,
262    /// The content of the message to be broadcasted, including specific data based on the message
263    /// type.
264    pub message: EthBroadcastMessage<N>,
265}
266
267impl<N: NetworkPrimitives> Encodable for ProtocolBroadcastMessage<N> {
268    /// Encodes the protocol message into bytes. The message type is encoded as a single byte and
269    /// prepended to the message.
270    fn encode(&self, out: &mut dyn BufMut) {
271        self.message_type.encode(out);
272        self.message.encode(out);
273    }
274    fn length(&self) -> usize {
275        self.message_type.length() + self.message.length()
276    }
277}
278
279impl<N: NetworkPrimitives> From<EthBroadcastMessage<N>> for ProtocolBroadcastMessage<N> {
280    fn from(message: EthBroadcastMessage<N>) -> Self {
281        Self { message_type: message.message_id(), message }
282    }
283}
284
285/// Represents a message in the eth wire protocol, versions 66, 67, 68 and 69.
286///
287/// The ethereum wire protocol is a set of messages that are broadcast to the network in two
288/// styles:
289///  * A request message sent by a peer (such as [`GetPooledTransactions`]), and an associated
290///    response message (such as [`PooledTransactions`]).
291///  * A message that is broadcast to the network, without a corresponding request.
292///
293/// The newer `eth/66` is an efficiency upgrade on top of `eth/65`, introducing a request id to
294/// correlate request-response message pairs. This allows for request multiplexing.
295///
296/// The `eth/67` is based on `eth/66` but only removes two messages, [`GetNodeData`] and
297/// [`NodeData`].
298///
299/// The `eth/68` changes only `NewPooledTransactionHashes` to include `types` and `sized`. For
300/// it, `NewPooledTransactionHashes` is renamed as [`NewPooledTransactionHashes66`] and
301/// [`NewPooledTransactionHashes68`] is defined.
302///
303/// The `eth/69` announces the historical block range served by the node. Removes total difficulty
304/// information. And removes the Bloom field from receipts transferred over the protocol.
305///
306/// The `eth/70` (EIP-7975) keeps the eth/69 status format and introduces partial receipts.
307/// requests/responses.
308///
309/// The `eth/71` draft extends eth/70 with block access list request/response messages.
310#[derive(Clone, Debug, PartialEq, Eq)]
311#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
312pub enum EthMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
313    /// Represents a Status message required for the protocol handshake.
314    Status(StatusMessage),
315    /// Represents a `NewBlockHashes` message broadcast to the network.
316    NewBlockHashes(NewBlockHashes),
317    /// Represents a `NewBlock` message broadcast to the network.
318    #[cfg_attr(
319        feature = "serde",
320        serde(bound = "N::NewBlockPayload: serde::Serialize + serde::de::DeserializeOwned")
321    )]
322    NewBlock(Box<N::NewBlockPayload>),
323    /// Represents a Transactions message broadcast to the network.
324    #[cfg_attr(
325        feature = "serde",
326        serde(bound = "N::BroadcastedTransaction: serde::Serialize + serde::de::DeserializeOwned")
327    )]
328    Transactions(Transactions<N::BroadcastedTransaction>),
329    /// Represents a `NewPooledTransactionHashes` message for eth/66 version.
330    NewPooledTransactionHashes66(NewPooledTransactionHashes66),
331    /// Represents a `NewPooledTransactionHashes` message for eth/68 version.
332    NewPooledTransactionHashes68(NewPooledTransactionHashes68),
333    /// Represents a `NewPooledTransactionHashes` message for eth/72 version.
334    NewPooledTransactionHashes72(NewPooledTransactionHashes72),
335
336    // The following messages are request-response message pairs
337    /// Represents a `GetBlockHeaders` request-response pair.
338    GetBlockHeaders(RequestPair<GetBlockHeaders>),
339    /// Represents a `BlockHeaders` request-response pair.
340    #[cfg_attr(
341        feature = "serde",
342        serde(bound = "N::BlockHeader: serde::Serialize + serde::de::DeserializeOwned")
343    )]
344    BlockHeaders(RequestPair<BlockHeaders<N::BlockHeader>>),
345    /// Represents a `GetBlockBodies` request-response pair.
346    GetBlockBodies(RequestPair<GetBlockBodies>),
347    /// Represents a `BlockBodies` request-response pair.
348    #[cfg_attr(
349        feature = "serde",
350        serde(bound = "N::BlockBody: serde::Serialize + serde::de::DeserializeOwned")
351    )]
352    BlockBodies(RequestPair<BlockBodies<N::BlockBody>>),
353    /// Represents a `GetPooledTransactions` request-response pair.
354    GetPooledTransactions(RequestPair<GetPooledTransactions>),
355    /// Represents a `PooledTransactions` request-response pair.
356    #[cfg_attr(
357        feature = "serde",
358        serde(bound = "N::PooledTransaction: serde::Serialize + serde::de::DeserializeOwned")
359    )]
360    PooledTransactions(RequestPair<PooledTransactions<N::PooledTransaction>>),
361    /// Represents a `GetNodeData` request-response pair.
362    GetNodeData(RequestPair<GetNodeData>),
363    /// Represents a `NodeData` request-response pair.
364    NodeData(RequestPair<NodeData>),
365    /// Represents a `GetReceipts` request-response pair.
366    GetReceipts(RequestPair<GetReceipts>),
367    /// Represents a `GetReceipts` request for eth/70.
368    ///
369    /// Note: Unlike earlier protocol versions, the eth/70 encoding for
370    /// `GetReceipts` in EIP-7975 inlines the request id. The type still wraps
371    /// a [`RequestPair`], but with a custom inline encoding.
372    GetReceipts70(RequestPair<GetReceipts70>),
373    /// Represents a `GetBlockAccessLists` request-response pair for eth/71.
374    GetBlockAccessLists(RequestPair<GetBlockAccessLists>),
375    /// Represents a Receipts request-response pair.
376    #[cfg_attr(
377        feature = "serde",
378        serde(bound = "N::Receipt: serde::Serialize + serde::de::DeserializeOwned")
379    )]
380    Receipts(RequestPair<Receipts<N::Receipt>>),
381    /// Represents a Receipts request-response pair for eth/69.
382    #[cfg_attr(
383        feature = "serde",
384        serde(bound = "N::Receipt: serde::Serialize + serde::de::DeserializeOwned")
385    )]
386    Receipts69(RequestPair<Receipts69<N::Receipt>>),
387    /// Represents a Receipts request-response pair for eth/70.
388    #[cfg_attr(
389        feature = "serde",
390        serde(bound = "N::Receipt: serde::Serialize + serde::de::DeserializeOwned")
391    )]
392    ///
393    /// Note: The eth/70 encoding for `Receipts` in EIP-7975 inlines the
394    /// request id. The type still wraps a [`RequestPair`], but with a custom
395    /// inline encoding.
396    Receipts70(RequestPair<Receipts70<N::Receipt>>),
397    /// Represents a `BlockAccessLists` request-response pair for eth/71.
398    BlockAccessLists(RequestPair<BlockAccessLists>),
399    /// Represents a `Cells` request-response pair for eth/72.
400    Cells(RequestPair<Cells>),
401    /// Represents a `GetCells` request-response pair for eth/72.
402    GetCells(RequestPair<GetCells>),
403    /// Represents a `BlockRangeUpdate` message broadcast to the network.
404    #[cfg_attr(
405        feature = "serde",
406        serde(bound = "N::BroadcastedTransaction: serde::Serialize + serde::de::DeserializeOwned")
407    )]
408    BlockRangeUpdate(BlockRangeUpdate),
409    /// Represents an encoded message that doesn't match any other variant
410    Other(RawCapabilityMessage),
411}
412
413impl<N: NetworkPrimitives> EthMessage<N> {
414    /// Returns the message's ID.
415    pub const fn message_id(&self) -> EthMessageID {
416        match self {
417            Self::Status(_) => EthMessageID::Status,
418            Self::NewBlockHashes(_) => EthMessageID::NewBlockHashes,
419            Self::NewBlock(_) => EthMessageID::NewBlock,
420            Self::Transactions(_) => EthMessageID::Transactions,
421            Self::NewPooledTransactionHashes66(_) |
422            Self::NewPooledTransactionHashes68(_) |
423            Self::NewPooledTransactionHashes72(_) => EthMessageID::NewPooledTransactionHashes,
424            Self::GetBlockHeaders(_) => EthMessageID::GetBlockHeaders,
425            Self::BlockHeaders(_) => EthMessageID::BlockHeaders,
426            Self::GetBlockBodies(_) => EthMessageID::GetBlockBodies,
427            Self::BlockBodies(_) => EthMessageID::BlockBodies,
428            Self::GetPooledTransactions(_) => EthMessageID::GetPooledTransactions,
429            Self::PooledTransactions(_) => EthMessageID::PooledTransactions,
430            Self::GetNodeData(_) => EthMessageID::GetNodeData,
431            Self::NodeData(_) => EthMessageID::NodeData,
432            Self::GetReceipts(_) | Self::GetReceipts70(_) => EthMessageID::GetReceipts,
433            Self::Receipts(_) | Self::Receipts69(_) | Self::Receipts70(_) => EthMessageID::Receipts,
434            Self::BlockRangeUpdate(_) => EthMessageID::BlockRangeUpdate,
435            Self::GetBlockAccessLists(_) => EthMessageID::GetBlockAccessLists,
436            Self::BlockAccessLists(_) => EthMessageID::BlockAccessLists,
437            Self::Cells(_) => EthMessageID::Cells,
438            Self::GetCells(_) => EthMessageID::GetCells,
439            Self::Other(msg) => EthMessageID::Other(msg.id as u8),
440        }
441    }
442
443    /// Returns true if the message variant is a request.
444    pub const fn is_request(&self) -> bool {
445        matches!(
446            self,
447            Self::GetBlockBodies(_) |
448                Self::GetBlockHeaders(_) |
449                Self::GetReceipts(_) |
450                Self::GetReceipts70(_) |
451                Self::GetBlockAccessLists(_) |
452                Self::GetCells(_) |
453                Self::GetPooledTransactions(_) |
454                Self::GetNodeData(_)
455        )
456    }
457
458    /// Returns true if the message variant is a response to a request.
459    pub const fn is_response(&self) -> bool {
460        matches!(
461            self,
462            Self::PooledTransactions(_) |
463                Self::Receipts(_) |
464                Self::Receipts69(_) |
465                Self::Receipts70(_) |
466                Self::BlockAccessLists(_) |
467                Self::BlockHeaders(_) |
468                Self::BlockBodies(_) |
469                Self::NodeData(_) |
470                Self::Cells(_)
471        )
472    }
473}
474
475impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
476    fn encode(&self, out: &mut dyn BufMut) {
477        match self {
478            Self::Status(status) => status.encode(out),
479            Self::NewBlockHashes(new_block_hashes) => new_block_hashes.encode(out),
480            Self::NewBlock(new_block) => new_block.encode(out),
481            Self::Transactions(transactions) => transactions.encode(out),
482            Self::NewPooledTransactionHashes66(hashes) => hashes.encode(out),
483            Self::NewPooledTransactionHashes68(hashes) => hashes.encode(out),
484            Self::NewPooledTransactionHashes72(hashes) => hashes.encode(out),
485            Self::GetBlockHeaders(request) => request.encode(out),
486            Self::BlockHeaders(headers) => headers.encode(out),
487            Self::GetBlockBodies(request) => request.encode(out),
488            Self::BlockBodies(bodies) => bodies.encode(out),
489            Self::GetPooledTransactions(request) => request.encode(out),
490            Self::PooledTransactions(transactions) => transactions.encode(out),
491            Self::GetNodeData(request) => request.encode(out),
492            Self::NodeData(data) => data.encode(out),
493            Self::GetReceipts(request) => request.encode(out),
494            Self::GetReceipts70(request) => request.encode(out),
495            Self::GetBlockAccessLists(request) => request.encode(out),
496            Self::GetCells(request) => request.encode(out),
497            Self::Receipts(receipts) => receipts.encode(out),
498            Self::Receipts69(receipt69) => receipt69.encode(out),
499            Self::Receipts70(receipt70) => receipt70.encode(out),
500            Self::BlockAccessLists(block_access_lists) => block_access_lists.encode(out),
501            Self::BlockRangeUpdate(block_range_update) => block_range_update.encode(out),
502            Self::Cells(cells) => cells.encode(out),
503            Self::Other(unknown) => out.put_slice(&unknown.payload),
504        }
505    }
506    fn length(&self) -> usize {
507        match self {
508            Self::Status(status) => status.length(),
509            Self::NewBlockHashes(new_block_hashes) => new_block_hashes.length(),
510            Self::NewBlock(new_block) => new_block.length(),
511            Self::Transactions(transactions) => transactions.length(),
512            Self::NewPooledTransactionHashes66(hashes) => hashes.length(),
513            Self::NewPooledTransactionHashes68(hashes) => hashes.length(),
514            Self::NewPooledTransactionHashes72(hashes) => hashes.length(),
515            Self::GetBlockHeaders(request) => request.length(),
516            Self::BlockHeaders(headers) => headers.length(),
517            Self::GetBlockBodies(request) => request.length(),
518            Self::BlockBodies(bodies) => bodies.length(),
519            Self::GetPooledTransactions(request) => request.length(),
520            Self::PooledTransactions(transactions) => transactions.length(),
521            Self::GetNodeData(request) => request.length(),
522            Self::NodeData(data) => data.length(),
523            Self::GetReceipts(request) => request.length(),
524            Self::GetReceipts70(request) => request.length(),
525            Self::GetBlockAccessLists(request) => request.length(),
526            Self::GetCells(request) => request.length(),
527            Self::Receipts(receipts) => receipts.length(),
528            Self::Receipts69(receipt69) => receipt69.length(),
529            Self::Receipts70(receipt70) => receipt70.length(),
530            Self::BlockAccessLists(block_access_lists) => block_access_lists.length(),
531            Self::BlockRangeUpdate(block_range_update) => block_range_update.length(),
532            Self::Cells(cells) => cells.length(),
533            Self::Other(unknown) => unknown.payload.len(),
534        }
535    }
536}
537
538/// Represents broadcast messages of [`EthMessage`] with the same object that can be sent to
539/// multiple peers.
540///
541/// Messages that contain a list of hashes depend on the peer the message is sent to. A peer should
542/// never receive a hash of an object (block, transaction) it has already seen.
543///
544/// Note: This is only useful for outgoing messages.
545#[derive(Clone, Debug)]
546pub enum EthBroadcastMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
547    /// Represents a new block broadcast message.
548    NewBlock(Arc<N::NewBlockPayload>),
549    /// Represents a transactions broadcast message.
550    Transactions(SharedTransactions<N::BroadcastedTransaction>),
551    /// Represents cached outbound pool transactions broadcast message.
552    BroadcastPoolTransactions(BroadcastPoolTransactions),
553}
554
555// === impl EthBroadcastMessage ===
556
557impl<N: NetworkPrimitives> EthBroadcastMessage<N> {
558    /// Returns the message's ID.
559    pub const fn message_id(&self) -> EthMessageID {
560        match self {
561            Self::NewBlock(_) => EthMessageID::NewBlock,
562            Self::Transactions(_) | Self::BroadcastPoolTransactions(_) => {
563                EthMessageID::Transactions
564            }
565        }
566    }
567
568    /// Encodes this broadcast to its id-prefixed `RLPx` message bytes.
569    pub fn encoded(self) -> alloy_primitives::bytes::Bytes {
570        alloy_rlp::encode(ProtocolBroadcastMessage::from(self)).into()
571    }
572}
573
574impl<N: NetworkPrimitives> Encodable for EthBroadcastMessage<N> {
575    fn encode(&self, out: &mut dyn BufMut) {
576        match self {
577            Self::NewBlock(new_block) => new_block.encode(out),
578            Self::Transactions(transactions) => transactions.encode(out),
579            Self::BroadcastPoolTransactions(transactions) => transactions.encode(out),
580        }
581    }
582
583    fn length(&self) -> usize {
584        match self {
585            Self::NewBlock(new_block) => new_block.length(),
586            Self::Transactions(transactions) => transactions.length(),
587            Self::BroadcastPoolTransactions(transactions) => transactions.length(),
588        }
589    }
590}
591
592/// Represents message IDs for eth protocol messages.
593#[repr(u8)]
594#[derive(Clone, Copy, Debug, PartialEq, Eq)]
595#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
596pub enum EthMessageID {
597    /// Status message.
598    Status = 0x00,
599    /// New block hashes message.
600    NewBlockHashes = 0x01,
601    /// Transactions message.
602    Transactions = 0x02,
603    /// Get block headers message.
604    GetBlockHeaders = 0x03,
605    /// Block headers message.
606    BlockHeaders = 0x04,
607    /// Get block bodies message.
608    GetBlockBodies = 0x05,
609    /// Block bodies message.
610    BlockBodies = 0x06,
611    /// New block message.
612    NewBlock = 0x07,
613    /// New pooled transaction hashes message.
614    NewPooledTransactionHashes = 0x08,
615    /// Requests pooled transactions.
616    GetPooledTransactions = 0x09,
617    /// Represents pooled transactions.
618    PooledTransactions = 0x0a,
619    /// Requests node data.
620    GetNodeData = 0x0d,
621    /// Represents node data.
622    NodeData = 0x0e,
623    /// Requests receipts.
624    GetReceipts = 0x0f,
625    /// Represents receipts.
626    Receipts = 0x10,
627    /// Block range update.
628    ///
629    /// Introduced in Eth69
630    BlockRangeUpdate = 0x11,
631    /// Requests block access lists.
632    ///
633    /// Introduced in Eth71
634    GetBlockAccessLists = 0x12,
635    /// Represents block access lists.
636    ///
637    /// Introduced in Eth71
638    BlockAccessLists = 0x13,
639
640    /// Requests cells.
641    ///
642    /// Introduced in Eth72
643    GetCells = 0x14,
644    /// Represents Cells
645    ///
646    /// Introduced in Eth72
647    Cells = 0x15,
648    /// Represents unknown message types.
649    Other(u8),
650}
651
652impl EthMessageID {
653    /// Returns the corresponding `u8` value for an `EthMessageID`.
654    pub const fn to_u8(&self) -> u8 {
655        match self {
656            Self::Status => 0x00,
657            Self::NewBlockHashes => 0x01,
658            Self::Transactions => 0x02,
659            Self::GetBlockHeaders => 0x03,
660            Self::BlockHeaders => 0x04,
661            Self::GetBlockBodies => 0x05,
662            Self::BlockBodies => 0x06,
663            Self::NewBlock => 0x07,
664            Self::NewPooledTransactionHashes => 0x08,
665            Self::GetPooledTransactions => 0x09,
666            Self::PooledTransactions => 0x0a,
667            Self::GetNodeData => 0x0d,
668            Self::NodeData => 0x0e,
669            Self::GetReceipts => 0x0f,
670            Self::Receipts => 0x10,
671            Self::BlockRangeUpdate => 0x11,
672            Self::GetBlockAccessLists => 0x12,
673            Self::BlockAccessLists => 0x13,
674            Self::GetCells => 0x14,
675            Self::Cells => 0x15,
676            Self::Other(value) => *value, // Return the stored `u8`
677        }
678    }
679
680    /// Returns the max value for the given version.
681    pub const fn max(version: EthVersion) -> u8 {
682        if version.is_eth72() {
683            Self::Cells.to_u8()
684        } else if version.is_eth71() {
685            Self::BlockAccessLists.to_u8()
686        } else if version.is_eth69_or_newer() {
687            Self::BlockRangeUpdate.to_u8()
688        } else {
689            Self::Receipts.to_u8()
690        }
691    }
692
693    /// Returns the total number of message types for the given version.
694    ///
695    /// This is used for message ID multiplexing.
696    ///
697    /// <https://github.com/ethereum/go-ethereum/blob/85077be58edea572f29c3b1a6a055077f1a56a8b/eth/protocols/eth/protocol.go#L45-L47>
698    pub const fn message_count(version: EthVersion) -> u8 {
699        Self::max(version) + 1
700    }
701}
702
703impl Encodable for EthMessageID {
704    fn encode(&self, out: &mut dyn BufMut) {
705        out.put_u8(self.to_u8());
706    }
707    fn length(&self) -> usize {
708        1
709    }
710}
711
712impl Decodable for EthMessageID {
713    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
714        let id = match buf.first().ok_or(alloy_rlp::Error::InputTooShort)? {
715            0x00 => Self::Status,
716            0x01 => Self::NewBlockHashes,
717            0x02 => Self::Transactions,
718            0x03 => Self::GetBlockHeaders,
719            0x04 => Self::BlockHeaders,
720            0x05 => Self::GetBlockBodies,
721            0x06 => Self::BlockBodies,
722            0x07 => Self::NewBlock,
723            0x08 => Self::NewPooledTransactionHashes,
724            0x09 => Self::GetPooledTransactions,
725            0x0a => Self::PooledTransactions,
726            0x0d => Self::GetNodeData,
727            0x0e => Self::NodeData,
728            0x0f => Self::GetReceipts,
729            0x10 => Self::Receipts,
730            0x11 => Self::BlockRangeUpdate,
731            0x12 => Self::GetBlockAccessLists,
732            0x13 => Self::BlockAccessLists,
733            0x14 => Self::GetCells,
734            0x15 => Self::Cells,
735            unknown => Self::Other(*unknown),
736        };
737        buf.advance(1);
738        Ok(id)
739    }
740}
741
742impl TryFrom<usize> for EthMessageID {
743    type Error = &'static str;
744
745    fn try_from(value: usize) -> Result<Self, Self::Error> {
746        match value {
747            0x00 => Ok(Self::Status),
748            0x01 => Ok(Self::NewBlockHashes),
749            0x02 => Ok(Self::Transactions),
750            0x03 => Ok(Self::GetBlockHeaders),
751            0x04 => Ok(Self::BlockHeaders),
752            0x05 => Ok(Self::GetBlockBodies),
753            0x06 => Ok(Self::BlockBodies),
754            0x07 => Ok(Self::NewBlock),
755            0x08 => Ok(Self::NewPooledTransactionHashes),
756            0x09 => Ok(Self::GetPooledTransactions),
757            0x0a => Ok(Self::PooledTransactions),
758            0x0d => Ok(Self::GetNodeData),
759            0x0e => Ok(Self::NodeData),
760            0x0f => Ok(Self::GetReceipts),
761            0x10 => Ok(Self::Receipts),
762            0x11 => Ok(Self::BlockRangeUpdate),
763            0x12 => Ok(Self::GetBlockAccessLists),
764            0x13 => Ok(Self::BlockAccessLists),
765            0x14 => Ok(Self::GetCells),
766            0x15 => Ok(Self::Cells),
767            _ => Err("Invalid message ID"),
768        }
769    }
770}
771
772/// This is used for all request-response style `eth` protocol messages.
773/// This can represent either a request or a response, since both include a message payload and
774/// request id.
775#[derive(Clone, Debug, PartialEq, Eq)]
776#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
777#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
778pub struct RequestPair<T> {
779    /// id for the contained request or response message
780    pub request_id: u64,
781
782    /// the request or response message payload
783    pub message: T,
784}
785
786impl<T> RequestPair<T> {
787    /// Converts the message type with the given closure.
788    pub fn map<F, R>(self, f: F) -> RequestPair<R>
789    where
790        F: FnOnce(T) -> R,
791    {
792        let Self { request_id, message } = self;
793        RequestPair { request_id, message: f(message) }
794    }
795
796    /// Decodes the request id and then decodes the message payload using `decode_msg`.
797    pub fn decode_with<F>(buf: &mut &[u8], decode_msg: F) -> alloy_rlp::Result<Self>
798    where
799        F: FnOnce(&mut &[u8]) -> alloy_rlp::Result<T>,
800    {
801        let header = Header::decode(buf)?;
802        if !header.list {
803            return Err(alloy_rlp::Error::UnexpectedString)
804        }
805
806        let initial_length = buf.len();
807        let request_id = u64::decode(buf)?;
808        let message = decode_msg(buf)?;
809
810        let consumed_len = initial_length - buf.len();
811        if consumed_len != header.payload_length {
812            return Err(alloy_rlp::Error::UnexpectedLength)
813        }
814
815        Ok(Self { request_id, message })
816    }
817}
818
819/// Allows messages with request ids to be serialized into RLP bytes.
820impl<T> Encodable for RequestPair<T>
821where
822    T: Encodable,
823{
824    fn encode(&self, out: &mut dyn alloy_rlp::BufMut) {
825        let header =
826            Header { list: true, payload_length: self.request_id.length() + self.message.length() };
827
828        header.encode(out);
829        self.request_id.encode(out);
830        self.message.encode(out);
831    }
832
833    fn length(&self) -> usize {
834        let payload_length = self.request_id.length() + self.message.length();
835        Header { list: true, payload_length }.length_with_payload()
836    }
837}
838
839/// Allows messages with request ids to be deserialized into RLP bytes.
840impl<T> Decodable for RequestPair<T>
841where
842    T: Decodable,
843{
844    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
845        let header = Header::decode(buf)?;
846        if !header.list {
847            return Err(alloy_rlp::Error::UnexpectedString)
848        }
849
850        let initial_length = buf.len();
851        let request_id = u64::decode(buf)?;
852        let message = T::decode(buf)?;
853
854        // Check that the buffer consumed exactly payload_length bytes after decoding the
855        // RequestPair
856        let consumed_len = initial_length - buf.len();
857        if consumed_len != header.payload_length {
858            return Err(alloy_rlp::Error::UnexpectedLength)
859        }
860
861        Ok(Self { request_id, message })
862    }
863}
864
865#[cfg(test)]
866mod tests {
867    use super::MessageError;
868    use crate::{
869        message::RequestPair, BlockAccessLists, EthMessage, EthMessageID, EthNetworkPrimitives,
870        EthVersion, GetBlockAccessLists, GetNodeData, NodeData, ProtocolMessage,
871        RawCapabilityMessage, Status, StatusMessage,
872    };
873    use alloy_hardforks::{ForkHash, ForkId};
874    use alloy_primitives::{hex, B256, U256};
875    use alloy_rlp::{Decodable, Encodable, Error};
876    use reth_ethereum_primitives::BlockBody;
877
878    fn encode<T: Encodable>(value: T) -> Vec<u8> {
879        let mut buf = vec![];
880        value.encode(&mut buf);
881        buf
882    }
883
884    #[test]
885    fn test_removed_message_at_eth67() {
886        let get_node_data = EthMessage::<EthNetworkPrimitives>::GetNodeData(RequestPair {
887            request_id: 1337,
888            message: GetNodeData(vec![]),
889        });
890        let buf = encode(ProtocolMessage {
891            message_type: EthMessageID::GetNodeData,
892            message: get_node_data,
893        });
894        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
895            crate::EthVersion::Eth67,
896            &mut &buf[..],
897        );
898        assert!(matches!(msg, Err(MessageError::Invalid(..))));
899
900        let node_data = EthMessage::<EthNetworkPrimitives>::NodeData(RequestPair {
901            request_id: 1337,
902            message: NodeData(vec![]),
903        });
904        let buf =
905            encode(ProtocolMessage { message_type: EthMessageID::NodeData, message: node_data });
906        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
907            crate::EthVersion::Eth67,
908            &mut &buf[..],
909        );
910        assert!(matches!(msg, Err(MessageError::Invalid(..))));
911    }
912
913    #[test]
914    fn test_bal_message_version_gating() {
915        let get_block_access_lists =
916            EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(RequestPair {
917                request_id: 1337,
918                message: GetBlockAccessLists(vec![]),
919            });
920        let buf = encode(ProtocolMessage {
921            message_type: EthMessageID::GetBlockAccessLists,
922            message: get_block_access_lists,
923        });
924        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
925            EthVersion::Eth70,
926            &mut &buf[..],
927        );
928        assert!(matches!(
929            msg,
930            Err(MessageError::Invalid(EthVersion::Eth70, EthMessageID::GetBlockAccessLists))
931        ));
932
933        let block_access_lists =
934            EthMessage::<EthNetworkPrimitives>::BlockAccessLists(RequestPair {
935                request_id: 1337,
936                message: BlockAccessLists(vec![]),
937            });
938        let buf = encode(ProtocolMessage {
939            message_type: EthMessageID::BlockAccessLists,
940            message: block_access_lists,
941        });
942        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
943            EthVersion::Eth70,
944            &mut &buf[..],
945        );
946        assert!(matches!(
947            msg,
948            Err(MessageError::Invalid(EthVersion::Eth70, EthMessageID::BlockAccessLists))
949        ));
950    }
951
952    #[test]
953    fn test_bal_message_eth71_roundtrip() {
954        let msg = ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(
955            RequestPair { request_id: 42, message: GetBlockAccessLists(vec![]) },
956        ));
957        let encoded = encode(msg.clone());
958        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
959            EthVersion::Eth71,
960            &mut &encoded[..],
961        )
962        .unwrap();
963
964        assert_eq!(decoded, msg);
965    }
966
967    #[test]
968    fn request_pair_encode() {
969        let request_pair = RequestPair { request_id: 1337, message: vec![5u8] };
970
971        // c5: start of list (c0) + len(full_list) (length is <55 bytes)
972        // 82: 0x80 + len(1337)
973        // 05 39: 1337 (request_id)
974        // === full_list ===
975        // c1: start of list (c0) + len(list) (length is <55 bytes)
976        // 05: 5 (message)
977        let expected = hex!("c5820539c105");
978        let got = encode(request_pair);
979        assert_eq!(expected[..], got, "expected: {expected:X?}, got: {got:X?}",);
980    }
981
982    #[test]
983    fn request_pair_decode() {
984        let raw_pair = &hex!("c5820539c105")[..];
985
986        let expected = RequestPair { request_id: 1337, message: vec![5u8] };
987
988        let got = RequestPair::<Vec<u8>>::decode(&mut &*raw_pair).unwrap();
989        assert_eq!(expected.length(), raw_pair.len());
990        assert_eq!(expected, got);
991    }
992
993    #[test]
994    fn malicious_request_pair_decode() {
995        // A maliciously encoded request pair, where the len(full_list) is 5, but it
996        // actually consumes 6 bytes when decoding
997        //
998        // c5: start of list (c0) + len(full_list) (length is <55 bytes)
999        // 82: 0x80 + len(1337)
1000        // 05 39: 1337 (request_id)
1001        // === full_list ===
1002        // c2: start of list (c0) + len(list) (length is <55 bytes)
1003        // 05 05: 5 5(message)
1004        let raw_pair = &hex!("c5820539c20505")[..];
1005
1006        let result = RequestPair::<Vec<u8>>::decode(&mut &*raw_pair);
1007        assert!(matches!(result, Err(Error::UnexpectedLength)));
1008    }
1009
1010    #[test]
1011    fn request_pair_decode_rejects_string_header() {
1012        // A request pair wrapped in a string header instead of a list header
1013        //
1014        // 82: start of string (80) + len(payload)
1015        // 2d: 45 (request_id)
1016        // c0: empty list (message)
1017        let raw_pair = &hex!("822dc0")[..];
1018
1019        let result = RequestPair::<Vec<u8>>::decode(&mut &*raw_pair);
1020        assert_eq!(result, Err(Error::UnexpectedString));
1021
1022        let result = RequestPair::decode_with(&mut &*raw_pair, Vec::<u8>::decode);
1023        assert_eq!(result, Err(Error::UnexpectedString));
1024    }
1025
1026    #[test]
1027    fn empty_block_bodies_protocol() {
1028        let empty_block_bodies =
1029            ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::BlockBodies(RequestPair {
1030                request_id: 0,
1031                message: Default::default(),
1032            }));
1033        let mut buf = Vec::new();
1034        empty_block_bodies.encode(&mut buf);
1035        let decoded =
1036            ProtocolMessage::decode_message(EthVersion::Eth68, &mut buf.as_slice()).unwrap();
1037        assert_eq!(empty_block_bodies, decoded);
1038    }
1039
1040    #[test]
1041    fn empty_block_body_protocol() {
1042        let empty_block_bodies =
1043            ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::BlockBodies(RequestPair {
1044                request_id: 0,
1045                message: vec![BlockBody {
1046                    transactions: vec![],
1047                    ommers: vec![],
1048                    withdrawals: Some(Default::default()),
1049                }]
1050                .into(),
1051            }));
1052        let mut buf = Vec::new();
1053        empty_block_bodies.encode(&mut buf);
1054        let decoded =
1055            ProtocolMessage::decode_message(EthVersion::Eth68, &mut buf.as_slice()).unwrap();
1056        assert_eq!(empty_block_bodies, decoded);
1057    }
1058
1059    #[test]
1060    fn decode_block_bodies_message() {
1061        let buf = hex!("06c48199c1c0");
1062        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
1063            EthVersion::Eth68,
1064            &mut &buf[..],
1065        )
1066        .unwrap_err();
1067        assert!(matches!(msg, MessageError::RlpError(alloy_rlp::Error::InputTooShort)));
1068    }
1069
1070    #[test]
1071    fn custom_message_roundtrip() {
1072        let custom_payload = vec![1, 2, 3, 4, 5];
1073        let custom_message = RawCapabilityMessage::new(0x20, custom_payload.into());
1074        let protocol_message = ProtocolMessage::<EthNetworkPrimitives> {
1075            message_type: EthMessageID::Other(0x20),
1076            message: EthMessage::Other(custom_message),
1077        };
1078
1079        let encoded = encode(protocol_message.clone());
1080        assert_eq!(protocol_message.length(), encoded.len());
1081        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
1082            EthVersion::Eth68,
1083            &mut &encoded[..],
1084        )
1085        .unwrap();
1086
1087        assert_eq!(protocol_message, decoded);
1088    }
1089
1090    #[test]
1091    fn custom_message_empty_payload_roundtrip() {
1092        let custom_message = RawCapabilityMessage::new(0x30, vec![].into());
1093        let protocol_message = ProtocolMessage::<EthNetworkPrimitives> {
1094            message_type: EthMessageID::Other(0x30),
1095            message: EthMessage::Other(custom_message),
1096        };
1097
1098        let encoded = encode(protocol_message.clone());
1099        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
1100            EthVersion::Eth68,
1101            &mut &encoded[..],
1102        )
1103        .unwrap();
1104
1105        assert_eq!(protocol_message, decoded);
1106    }
1107
1108    #[test]
1109    fn decode_status_success() {
1110        let status = Status {
1111            version: EthVersion::Eth68,
1112            chain: alloy_chains::Chain::mainnet(),
1113            total_difficulty: U256::from(100u64),
1114            blockhash: B256::random(),
1115            genesis: B256::random(),
1116            forkid: ForkId { hash: ForkHash([0xb7, 0x15, 0x07, 0x7d]), next: 0 },
1117        };
1118
1119        let protocol_message = ProtocolMessage::<EthNetworkPrimitives>::from(EthMessage::Status(
1120            StatusMessage::Legacy(status),
1121        ));
1122        let encoded = encode(protocol_message);
1123
1124        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_status(
1125            EthVersion::Eth68,
1126            &mut &encoded[..],
1127        )
1128        .unwrap();
1129
1130        assert!(matches!(decoded, StatusMessage::Legacy(s) if s == status));
1131    }
1132
1133    #[test]
1134    fn eth_message_id_max_includes_block_range_update() {
1135        assert_eq!(EthMessageID::max(EthVersion::Eth69), EthMessageID::BlockRangeUpdate.to_u8(),);
1136        assert_eq!(EthMessageID::max(EthVersion::Eth70), EthMessageID::BlockRangeUpdate.to_u8(),);
1137        assert_eq!(EthMessageID::max(EthVersion::Eth68), EthMessageID::Receipts.to_u8());
1138    }
1139
1140    #[test]
1141    fn decode_status_rejects_non_status() {
1142        let msg = EthMessage::<EthNetworkPrimitives>::GetBlockBodies(RequestPair {
1143            request_id: 1,
1144            message: crate::GetBlockBodies::default(),
1145        });
1146        let protocol_message =
1147            ProtocolMessage { message_type: EthMessageID::GetBlockBodies, message: msg };
1148        let encoded = encode(protocol_message);
1149
1150        let result = ProtocolMessage::<EthNetworkPrimitives>::decode_status(
1151            EthVersion::Eth68,
1152            &mut &encoded[..],
1153        );
1154
1155        assert!(matches!(
1156            result,
1157            Err(MessageError::ExpectedStatusMessage(EthMessageID::GetBlockBodies))
1158        ));
1159    }
1160}
reth_eth_wire_types/message.rs

reth_eth_wire_types/
message.rs
