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::{length_of_length, 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    /// Converts the message types where applicable.
475    ///
476    /// This handles up/downcasting where appropriate, for example for different receipt request
477    /// types.
478    pub fn map_versioned(self, version: EthVersion) -> Self {
479        // For eth/70 peers we send `GetReceipts` using the new eth/70
480        // encoding with `firstBlockReceiptIndex = 0`, while keeping the
481        // user-facing `PeerRequest` API unchanged.
482        if version >= EthVersion::Eth70 {
483            return match self {
484                Self::GetReceipts(pair) => {
485                    let RequestPair { request_id, message } = pair;
486                    let req = RequestPair {
487                        request_id,
488                        message: GetReceipts70 {
489                            first_block_receipt_index: 0,
490                            block_hashes: message.0,
491                        },
492                    };
493                    Self::GetReceipts70(req)
494                }
495                other => other,
496            }
497        }
498
499        self
500    }
501}
502
503impl<N: NetworkPrimitives> Encodable for EthMessage<N> {
504    fn encode(&self, out: &mut dyn BufMut) {
505        match self {
506            Self::Status(status) => status.encode(out),
507            Self::NewBlockHashes(new_block_hashes) => new_block_hashes.encode(out),
508            Self::NewBlock(new_block) => new_block.encode(out),
509            Self::Transactions(transactions) => transactions.encode(out),
510            Self::NewPooledTransactionHashes66(hashes) => hashes.encode(out),
511            Self::NewPooledTransactionHashes68(hashes) => hashes.encode(out),
512            Self::NewPooledTransactionHashes72(hashes) => hashes.encode(out),
513            Self::GetBlockHeaders(request) => request.encode(out),
514            Self::BlockHeaders(headers) => headers.encode(out),
515            Self::GetBlockBodies(request) => request.encode(out),
516            Self::BlockBodies(bodies) => bodies.encode(out),
517            Self::GetPooledTransactions(request) => request.encode(out),
518            Self::PooledTransactions(transactions) => transactions.encode(out),
519            Self::GetNodeData(request) => request.encode(out),
520            Self::NodeData(data) => data.encode(out),
521            Self::GetReceipts(request) => request.encode(out),
522            Self::GetReceipts70(request) => request.encode(out),
523            Self::GetBlockAccessLists(request) => request.encode(out),
524            Self::GetCells(request) => request.encode(out),
525            Self::Receipts(receipts) => receipts.encode(out),
526            Self::Receipts69(receipt69) => receipt69.encode(out),
527            Self::Receipts70(receipt70) => receipt70.encode(out),
528            Self::BlockAccessLists(block_access_lists) => block_access_lists.encode(out),
529            Self::BlockRangeUpdate(block_range_update) => block_range_update.encode(out),
530            Self::Cells(cells) => cells.encode(out),
531            Self::Other(unknown) => out.put_slice(&unknown.payload),
532        }
533    }
534    fn length(&self) -> usize {
535        match self {
536            Self::Status(status) => status.length(),
537            Self::NewBlockHashes(new_block_hashes) => new_block_hashes.length(),
538            Self::NewBlock(new_block) => new_block.length(),
539            Self::Transactions(transactions) => transactions.length(),
540            Self::NewPooledTransactionHashes66(hashes) => hashes.length(),
541            Self::NewPooledTransactionHashes68(hashes) => hashes.length(),
542            Self::NewPooledTransactionHashes72(hashes) => hashes.length(),
543            Self::GetBlockHeaders(request) => request.length(),
544            Self::BlockHeaders(headers) => headers.length(),
545            Self::GetBlockBodies(request) => request.length(),
546            Self::BlockBodies(bodies) => bodies.length(),
547            Self::GetPooledTransactions(request) => request.length(),
548            Self::PooledTransactions(transactions) => transactions.length(),
549            Self::GetNodeData(request) => request.length(),
550            Self::NodeData(data) => data.length(),
551            Self::GetReceipts(request) => request.length(),
552            Self::GetReceipts70(request) => request.length(),
553            Self::GetBlockAccessLists(request) => request.length(),
554            Self::GetCells(request) => request.length(),
555            Self::Receipts(receipts) => receipts.length(),
556            Self::Receipts69(receipt69) => receipt69.length(),
557            Self::Receipts70(receipt70) => receipt70.length(),
558            Self::BlockAccessLists(block_access_lists) => block_access_lists.length(),
559            Self::BlockRangeUpdate(block_range_update) => block_range_update.length(),
560            Self::Cells(cells) => cells.length(),
561            Self::Other(unknown) => unknown.length(),
562        }
563    }
564}
565
566/// Represents broadcast messages of [`EthMessage`] with the same object that can be sent to
567/// multiple peers.
568///
569/// Messages that contain a list of hashes depend on the peer the message is sent to. A peer should
570/// never receive a hash of an object (block, transaction) it has already seen.
571///
572/// Note: This is only useful for outgoing messages.
573#[derive(Clone, Debug)]
574pub enum EthBroadcastMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
575    /// Represents a new block broadcast message.
576    NewBlock(Arc<N::NewBlockPayload>),
577    /// Represents a transactions broadcast message.
578    Transactions(SharedTransactions<N::BroadcastedTransaction>),
579    /// Represents cached outbound pool transactions broadcast message.
580    BroadcastPoolTransactions(BroadcastPoolTransactions),
581}
582
583// === impl EthBroadcastMessage ===
584
585impl<N: NetworkPrimitives> EthBroadcastMessage<N> {
586    /// Returns the message's ID.
587    pub const fn message_id(&self) -> EthMessageID {
588        match self {
589            Self::NewBlock(_) => EthMessageID::NewBlock,
590            Self::Transactions(_) | Self::BroadcastPoolTransactions(_) => {
591                EthMessageID::Transactions
592            }
593        }
594    }
595
596    /// Encodes this broadcast to its id-prefixed `RLPx` message bytes.
597    pub fn encoded(self) -> alloy_primitives::bytes::Bytes {
598        alloy_rlp::encode(ProtocolBroadcastMessage::from(self)).into()
599    }
600}
601
602impl<N: NetworkPrimitives> Encodable for EthBroadcastMessage<N> {
603    fn encode(&self, out: &mut dyn BufMut) {
604        match self {
605            Self::NewBlock(new_block) => new_block.encode(out),
606            Self::Transactions(transactions) => transactions.encode(out),
607            Self::BroadcastPoolTransactions(transactions) => transactions.encode(out),
608        }
609    }
610
611    fn length(&self) -> usize {
612        match self {
613            Self::NewBlock(new_block) => new_block.length(),
614            Self::Transactions(transactions) => transactions.length(),
615            Self::BroadcastPoolTransactions(transactions) => transactions.length(),
616        }
617    }
618}
619
620/// Represents message IDs for eth protocol messages.
621#[repr(u8)]
622#[derive(Clone, Copy, Debug, PartialEq, Eq)]
623#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
624pub enum EthMessageID {
625    /// Status message.
626    Status = 0x00,
627    /// New block hashes message.
628    NewBlockHashes = 0x01,
629    /// Transactions message.
630    Transactions = 0x02,
631    /// Get block headers message.
632    GetBlockHeaders = 0x03,
633    /// Block headers message.
634    BlockHeaders = 0x04,
635    /// Get block bodies message.
636    GetBlockBodies = 0x05,
637    /// Block bodies message.
638    BlockBodies = 0x06,
639    /// New block message.
640    NewBlock = 0x07,
641    /// New pooled transaction hashes message.
642    NewPooledTransactionHashes = 0x08,
643    /// Requests pooled transactions.
644    GetPooledTransactions = 0x09,
645    /// Represents pooled transactions.
646    PooledTransactions = 0x0a,
647    /// Requests node data.
648    GetNodeData = 0x0d,
649    /// Represents node data.
650    NodeData = 0x0e,
651    /// Requests receipts.
652    GetReceipts = 0x0f,
653    /// Represents receipts.
654    Receipts = 0x10,
655    /// Block range update.
656    ///
657    /// Introduced in Eth69
658    BlockRangeUpdate = 0x11,
659    /// Requests block access lists.
660    ///
661    /// Introduced in Eth71
662    GetBlockAccessLists = 0x12,
663    /// Represents block access lists.
664    ///
665    /// Introduced in Eth71
666    BlockAccessLists = 0x13,
667
668    /// Requests cells.
669    ///
670    /// Introduced in Eth72
671    GetCells = 0x14,
672    /// Represents Cells
673    ///
674    /// Introduced in Eth72
675    Cells = 0x15,
676    /// Represents unknown message types.
677    Other(u8),
678}
679
680impl EthMessageID {
681    /// Returns the corresponding `u8` value for an `EthMessageID`.
682    pub const fn to_u8(&self) -> u8 {
683        match self {
684            Self::Status => 0x00,
685            Self::NewBlockHashes => 0x01,
686            Self::Transactions => 0x02,
687            Self::GetBlockHeaders => 0x03,
688            Self::BlockHeaders => 0x04,
689            Self::GetBlockBodies => 0x05,
690            Self::BlockBodies => 0x06,
691            Self::NewBlock => 0x07,
692            Self::NewPooledTransactionHashes => 0x08,
693            Self::GetPooledTransactions => 0x09,
694            Self::PooledTransactions => 0x0a,
695            Self::GetNodeData => 0x0d,
696            Self::NodeData => 0x0e,
697            Self::GetReceipts => 0x0f,
698            Self::Receipts => 0x10,
699            Self::BlockRangeUpdate => 0x11,
700            Self::GetBlockAccessLists => 0x12,
701            Self::BlockAccessLists => 0x13,
702            Self::GetCells => 0x14,
703            Self::Cells => 0x15,
704            Self::Other(value) => *value, // Return the stored `u8`
705        }
706    }
707
708    /// Returns the max value for the given version.
709    pub const fn max(version: EthVersion) -> u8 {
710        if version.is_eth72() {
711            Self::Cells.to_u8()
712        } else if version.is_eth71() {
713            Self::BlockAccessLists.to_u8()
714        } else if version.is_eth69_or_newer() {
715            Self::BlockRangeUpdate.to_u8()
716        } else {
717            Self::Receipts.to_u8()
718        }
719    }
720
721    /// Returns the total number of message types for the given version.
722    ///
723    /// This is used for message ID multiplexing.
724    ///
725    /// <https://github.com/ethereum/go-ethereum/blob/85077be58edea572f29c3b1a6a055077f1a56a8b/eth/protocols/eth/protocol.go#L45-L47>
726    pub const fn message_count(version: EthVersion) -> u8 {
727        Self::max(version) + 1
728    }
729}
730
731impl Encodable for EthMessageID {
732    fn encode(&self, out: &mut dyn BufMut) {
733        out.put_u8(self.to_u8());
734    }
735    fn length(&self) -> usize {
736        1
737    }
738}
739
740impl Decodable for EthMessageID {
741    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
742        let id = match buf.first().ok_or(alloy_rlp::Error::InputTooShort)? {
743            0x00 => Self::Status,
744            0x01 => Self::NewBlockHashes,
745            0x02 => Self::Transactions,
746            0x03 => Self::GetBlockHeaders,
747            0x04 => Self::BlockHeaders,
748            0x05 => Self::GetBlockBodies,
749            0x06 => Self::BlockBodies,
750            0x07 => Self::NewBlock,
751            0x08 => Self::NewPooledTransactionHashes,
752            0x09 => Self::GetPooledTransactions,
753            0x0a => Self::PooledTransactions,
754            0x0d => Self::GetNodeData,
755            0x0e => Self::NodeData,
756            0x0f => Self::GetReceipts,
757            0x10 => Self::Receipts,
758            0x11 => Self::BlockRangeUpdate,
759            0x12 => Self::GetBlockAccessLists,
760            0x13 => Self::BlockAccessLists,
761            0x14 => Self::GetCells,
762            0x15 => Self::Cells,
763            unknown => Self::Other(*unknown),
764        };
765        buf.advance(1);
766        Ok(id)
767    }
768}
769
770impl TryFrom<usize> for EthMessageID {
771    type Error = &'static str;
772
773    fn try_from(value: usize) -> Result<Self, Self::Error> {
774        match value {
775            0x00 => Ok(Self::Status),
776            0x01 => Ok(Self::NewBlockHashes),
777            0x02 => Ok(Self::Transactions),
778            0x03 => Ok(Self::GetBlockHeaders),
779            0x04 => Ok(Self::BlockHeaders),
780            0x05 => Ok(Self::GetBlockBodies),
781            0x06 => Ok(Self::BlockBodies),
782            0x07 => Ok(Self::NewBlock),
783            0x08 => Ok(Self::NewPooledTransactionHashes),
784            0x09 => Ok(Self::GetPooledTransactions),
785            0x0a => Ok(Self::PooledTransactions),
786            0x0d => Ok(Self::GetNodeData),
787            0x0e => Ok(Self::NodeData),
788            0x0f => Ok(Self::GetReceipts),
789            0x10 => Ok(Self::Receipts),
790            0x11 => Ok(Self::BlockRangeUpdate),
791            0x12 => Ok(Self::GetBlockAccessLists),
792            0x13 => Ok(Self::BlockAccessLists),
793            0x14 => Ok(Self::GetCells),
794            0x15 => Ok(Self::Cells),
795            _ => Err("Invalid message ID"),
796        }
797    }
798}
799
800/// This is used for all request-response style `eth` protocol messages.
801/// This can represent either a request or a response, since both include a message payload and
802/// request id.
803#[derive(Clone, Debug, PartialEq, Eq)]
804#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
805#[cfg_attr(any(test, feature = "arbitrary"), derive(arbitrary::Arbitrary))]
806pub struct RequestPair<T> {
807    /// id for the contained request or response message
808    pub request_id: u64,
809
810    /// the request or response message payload
811    pub message: T,
812}
813
814impl<T> RequestPair<T> {
815    /// Converts the message type with the given closure.
816    pub fn map<F, R>(self, f: F) -> RequestPair<R>
817    where
818        F: FnOnce(T) -> R,
819    {
820        let Self { request_id, message } = self;
821        RequestPair { request_id, message: f(message) }
822    }
823
824    /// Decodes the request id and then decodes the message payload using `decode_msg`.
825    pub fn decode_with<F>(buf: &mut &[u8], decode_msg: F) -> alloy_rlp::Result<Self>
826    where
827        F: FnOnce(&mut &[u8]) -> alloy_rlp::Result<T>,
828    {
829        let header = Header::decode(buf)?;
830
831        let initial_length = buf.len();
832        let request_id = u64::decode(buf)?;
833        let message = decode_msg(buf)?;
834
835        let consumed_len = initial_length - buf.len();
836        if consumed_len != header.payload_length {
837            return Err(alloy_rlp::Error::UnexpectedLength)
838        }
839
840        Ok(Self { request_id, message })
841    }
842}
843
844/// Allows messages with request ids to be serialized into RLP bytes.
845impl<T> Encodable for RequestPair<T>
846where
847    T: Encodable,
848{
849    fn encode(&self, out: &mut dyn alloy_rlp::BufMut) {
850        let header =
851            Header { list: true, payload_length: self.request_id.length() + self.message.length() };
852
853        header.encode(out);
854        self.request_id.encode(out);
855        self.message.encode(out);
856    }
857
858    fn length(&self) -> usize {
859        let mut length = 0;
860        length += self.request_id.length();
861        length += self.message.length();
862        length += length_of_length(length);
863        length
864    }
865}
866
867/// Allows messages with request ids to be deserialized into RLP bytes.
868impl<T> Decodable for RequestPair<T>
869where
870    T: Decodable,
871{
872    fn decode(buf: &mut &[u8]) -> alloy_rlp::Result<Self> {
873        let header = Header::decode(buf)?;
874
875        let initial_length = buf.len();
876        let request_id = u64::decode(buf)?;
877        let message = T::decode(buf)?;
878
879        // Check that the buffer consumed exactly payload_length bytes after decoding the
880        // RequestPair
881        let consumed_len = initial_length - buf.len();
882        if consumed_len != header.payload_length {
883            return Err(alloy_rlp::Error::UnexpectedLength)
884        }
885
886        Ok(Self { request_id, message })
887    }
888}
889
890#[cfg(test)]
891mod tests {
892    use super::MessageError;
893    use crate::{
894        message::RequestPair, BlockAccessLists, EthMessage, EthMessageID, EthNetworkPrimitives,
895        EthVersion, GetBlockAccessLists, GetNodeData, NodeData, ProtocolMessage,
896        RawCapabilityMessage,
897    };
898    use alloy_primitives::hex;
899    use alloy_rlp::{Decodable, Encodable, Error};
900    use reth_ethereum_primitives::BlockBody;
901
902    fn encode<T: Encodable>(value: T) -> Vec<u8> {
903        let mut buf = vec![];
904        value.encode(&mut buf);
905        buf
906    }
907
908    #[test]
909    fn test_removed_message_at_eth67() {
910        let get_node_data = EthMessage::<EthNetworkPrimitives>::GetNodeData(RequestPair {
911            request_id: 1337,
912            message: GetNodeData(vec![]),
913        });
914        let buf = encode(ProtocolMessage {
915            message_type: EthMessageID::GetNodeData,
916            message: get_node_data,
917        });
918        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
919            crate::EthVersion::Eth67,
920            &mut &buf[..],
921        );
922        assert!(matches!(msg, Err(MessageError::Invalid(..))));
923
924        let node_data = EthMessage::<EthNetworkPrimitives>::NodeData(RequestPair {
925            request_id: 1337,
926            message: NodeData(vec![]),
927        });
928        let buf =
929            encode(ProtocolMessage { message_type: EthMessageID::NodeData, message: node_data });
930        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
931            crate::EthVersion::Eth67,
932            &mut &buf[..],
933        );
934        assert!(matches!(msg, Err(MessageError::Invalid(..))));
935    }
936
937    #[test]
938    fn test_bal_message_version_gating() {
939        let get_block_access_lists =
940            EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(RequestPair {
941                request_id: 1337,
942                message: GetBlockAccessLists(vec![]),
943            });
944        let buf = encode(ProtocolMessage {
945            message_type: EthMessageID::GetBlockAccessLists,
946            message: get_block_access_lists,
947        });
948        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
949            EthVersion::Eth70,
950            &mut &buf[..],
951        );
952        assert!(matches!(
953            msg,
954            Err(MessageError::Invalid(EthVersion::Eth70, EthMessageID::GetBlockAccessLists))
955        ));
956
957        let block_access_lists =
958            EthMessage::<EthNetworkPrimitives>::BlockAccessLists(RequestPair {
959                request_id: 1337,
960                message: BlockAccessLists(vec![]),
961            });
962        let buf = encode(ProtocolMessage {
963            message_type: EthMessageID::BlockAccessLists,
964            message: block_access_lists,
965        });
966        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
967            EthVersion::Eth70,
968            &mut &buf[..],
969        );
970        assert!(matches!(
971            msg,
972            Err(MessageError::Invalid(EthVersion::Eth70, EthMessageID::BlockAccessLists))
973        ));
974    }
975
976    #[test]
977    fn test_bal_message_eth71_roundtrip() {
978        let msg = ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::GetBlockAccessLists(
979            RequestPair { request_id: 42, message: GetBlockAccessLists(vec![]) },
980        ));
981        let encoded = encode(msg.clone());
982        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
983            EthVersion::Eth71,
984            &mut &encoded[..],
985        )
986        .unwrap();
987
988        assert_eq!(decoded, msg);
989    }
990
991    #[test]
992    fn request_pair_encode() {
993        let request_pair = RequestPair { request_id: 1337, message: vec![5u8] };
994
995        // c5: start of list (c0) + len(full_list) (length is <55 bytes)
996        // 82: 0x80 + len(1337)
997        // 05 39: 1337 (request_id)
998        // === full_list ===
999        // c1: start of list (c0) + len(list) (length is <55 bytes)
1000        // 05: 5 (message)
1001        let expected = hex!("c5820539c105");
1002        let got = encode(request_pair);
1003        assert_eq!(expected[..], got, "expected: {expected:X?}, got: {got:X?}",);
1004    }
1005
1006    #[test]
1007    fn request_pair_decode() {
1008        let raw_pair = &hex!("c5820539c105")[..];
1009
1010        let expected = RequestPair { request_id: 1337, message: vec![5u8] };
1011
1012        let got = RequestPair::<Vec<u8>>::decode(&mut &*raw_pair).unwrap();
1013        assert_eq!(expected.length(), raw_pair.len());
1014        assert_eq!(expected, got);
1015    }
1016
1017    #[test]
1018    fn malicious_request_pair_decode() {
1019        // A maliciously encoded request pair, where the len(full_list) is 5, but it
1020        // actually consumes 6 bytes when decoding
1021        //
1022        // c5: start of list (c0) + len(full_list) (length is <55 bytes)
1023        // 82: 0x80 + len(1337)
1024        // 05 39: 1337 (request_id)
1025        // === full_list ===
1026        // c2: start of list (c0) + len(list) (length is <55 bytes)
1027        // 05 05: 5 5(message)
1028        let raw_pair = &hex!("c5820539c20505")[..];
1029
1030        let result = RequestPair::<Vec<u8>>::decode(&mut &*raw_pair);
1031        assert!(matches!(result, Err(Error::UnexpectedLength)));
1032    }
1033
1034    #[test]
1035    fn empty_block_bodies_protocol() {
1036        let empty_block_bodies =
1037            ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::BlockBodies(RequestPair {
1038                request_id: 0,
1039                message: Default::default(),
1040            }));
1041        let mut buf = Vec::new();
1042        empty_block_bodies.encode(&mut buf);
1043        let decoded =
1044            ProtocolMessage::decode_message(EthVersion::Eth68, &mut buf.as_slice()).unwrap();
1045        assert_eq!(empty_block_bodies, decoded);
1046    }
1047
1048    #[test]
1049    fn empty_block_body_protocol() {
1050        let empty_block_bodies =
1051            ProtocolMessage::from(EthMessage::<EthNetworkPrimitives>::BlockBodies(RequestPair {
1052                request_id: 0,
1053                message: vec![BlockBody {
1054                    transactions: vec![],
1055                    ommers: vec![],
1056                    withdrawals: Some(Default::default()),
1057                }]
1058                .into(),
1059            }));
1060        let mut buf = Vec::new();
1061        empty_block_bodies.encode(&mut buf);
1062        let decoded =
1063            ProtocolMessage::decode_message(EthVersion::Eth68, &mut buf.as_slice()).unwrap();
1064        assert_eq!(empty_block_bodies, decoded);
1065    }
1066
1067    #[test]
1068    fn decode_block_bodies_message() {
1069        let buf = hex!("06c48199c1c0");
1070        let msg = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
1071            EthVersion::Eth68,
1072            &mut &buf[..],
1073        )
1074        .unwrap_err();
1075        assert!(matches!(msg, MessageError::RlpError(alloy_rlp::Error::InputTooShort)));
1076    }
1077
1078    #[test]
1079    fn custom_message_roundtrip() {
1080        let custom_payload = vec![1, 2, 3, 4, 5];
1081        let custom_message = RawCapabilityMessage::new(0x20, custom_payload.into());
1082        let protocol_message = ProtocolMessage::<EthNetworkPrimitives> {
1083            message_type: EthMessageID::Other(0x20),
1084            message: EthMessage::Other(custom_message),
1085        };
1086
1087        let encoded = encode(protocol_message.clone());
1088        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
1089            EthVersion::Eth68,
1090            &mut &encoded[..],
1091        )
1092        .unwrap();
1093
1094        assert_eq!(protocol_message, decoded);
1095    }
1096
1097    #[test]
1098    fn custom_message_empty_payload_roundtrip() {
1099        let custom_message = RawCapabilityMessage::new(0x30, vec![].into());
1100        let protocol_message = ProtocolMessage::<EthNetworkPrimitives> {
1101            message_type: EthMessageID::Other(0x30),
1102            message: EthMessage::Other(custom_message),
1103        };
1104
1105        let encoded = encode(protocol_message.clone());
1106        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_message(
1107            EthVersion::Eth68,
1108            &mut &encoded[..],
1109        )
1110        .unwrap();
1111
1112        assert_eq!(protocol_message, decoded);
1113    }
1114
1115    #[test]
1116    fn decode_status_success() {
1117        use crate::{Status, StatusMessage};
1118        use alloy_hardforks::{ForkHash, ForkId};
1119        use alloy_primitives::{B256, U256};
1120
1121        let status = Status {
1122            version: EthVersion::Eth68,
1123            chain: alloy_chains::Chain::mainnet(),
1124            total_difficulty: U256::from(100u64),
1125            blockhash: B256::random(),
1126            genesis: B256::random(),
1127            forkid: ForkId { hash: ForkHash([0xb7, 0x15, 0x07, 0x7d]), next: 0 },
1128        };
1129
1130        let protocol_message = ProtocolMessage::<EthNetworkPrimitives>::from(EthMessage::Status(
1131            StatusMessage::Legacy(status),
1132        ));
1133        let encoded = encode(protocol_message);
1134
1135        let decoded = ProtocolMessage::<EthNetworkPrimitives>::decode_status(
1136            EthVersion::Eth68,
1137            &mut &encoded[..],
1138        )
1139        .unwrap();
1140
1141        assert!(matches!(decoded, StatusMessage::Legacy(s) if s == status));
1142    }
1143
1144    #[test]
1145    fn eth_message_id_max_includes_block_range_update() {
1146        assert_eq!(EthMessageID::max(EthVersion::Eth69), EthMessageID::BlockRangeUpdate.to_u8(),);
1147        assert_eq!(EthMessageID::max(EthVersion::Eth70), EthMessageID::BlockRangeUpdate.to_u8(),);
1148        assert_eq!(EthMessageID::max(EthVersion::Eth68), EthMessageID::Receipts.to_u8());
1149    }
1150
1151    #[test]
1152    fn decode_status_rejects_non_status() {
1153        let msg = EthMessage::<EthNetworkPrimitives>::GetBlockBodies(RequestPair {
1154            request_id: 1,
1155            message: crate::GetBlockBodies::default(),
1156        });
1157        let protocol_message =
1158            ProtocolMessage { message_type: EthMessageID::GetBlockBodies, message: msg };
1159        let encoded = encode(protocol_message);
1160
1161        let result = ProtocolMessage::<EthNetworkPrimitives>::decode_status(
1162            EthVersion::Eth68,
1163            &mut &encoded[..],
1164        );
1165
1166        assert!(matches!(
1167            result,
1168            Err(MessageError::ExpectedStatusMessage(EthMessageID::GetBlockBodies))
1169        ));
1170    }
1171}
reth_eth_wire_types/message.rs

reth_eth_wire_types/
message.rs
