reth_network/
message.rs

1//! Capability messaging
2//!
3//! An `RLPx` stream is multiplexed via the prepended message-id of a framed message.
4//! Capabilities are exchanged via the `RLPx` `Hello` message as pairs of `(id, version)`, <https://github.com/ethereum/devp2p/blob/master/rlpx.md#capability-messaging>
5
6use crate::types::{BlockAccessLists, Receipts69, Receipts70};
7use alloy_consensus::{BlockHeader, ReceiptWithBloom};
8use alloy_primitives::{Bytes, B256};
9use futures::FutureExt;
10use reth_eth_wire::{
11    message::RequestPair, BlockBodies, BlockHeaders, BlockRangeUpdate, BroadcastPoolTransactions,
12    Cells, EthMessage, EthNetworkPrimitives, GetBlockAccessLists, GetBlockBodies, GetBlockHeaders,
13    GetReceipts, NetworkPrimitives, NewBlock, NewBlockHashes, NewBlockPayload,
14    NewPooledTransactionHashes, NodeData, PooledTransactions, Receipts, SharedTransactions,
15    Transactions,
16};
17use reth_eth_wire_types::{snap::SnapProtocolMessage, RawCapabilityMessage};
18use reth_network_api::{PeerRequest, RequestMessage};
19use reth_network_p2p::{
20    error::{RequestError, RequestResult},
21    snap::client::SnapResponse,
22};
23use reth_primitives_traits::Block;
24use std::{
25    sync::Arc,
26    task::{ready, Context, Poll},
27};
28use tokio::sync::oneshot;
29
30/// Internal form of a `NewBlock` message
31#[derive(Debug, Clone)]
32pub struct NewBlockMessage<P = NewBlock<reth_ethereum_primitives::Block>> {
33    /// Hash of the block
34    pub hash: B256,
35    /// Raw received message
36    pub block: Arc<P>,
37}
38
39// === impl NewBlockMessage ===
40
41impl<P: NewBlockPayload> NewBlockMessage<P> {
42    /// Returns the block number of the block
43    pub fn number(&self) -> u64 {
44        self.block.block().header().number()
45    }
46}
47
48/// All Bi-directional eth-message variants that can be sent to a session or received from a
49/// session.
50#[derive(Debug)]
51pub enum PeerMessage<N: NetworkPrimitives = EthNetworkPrimitives> {
52    /// Announce new block hashes
53    NewBlockHashes(NewBlockHashes),
54    /// Broadcast new block.
55    NewBlock(NewBlockMessage<N::NewBlockPayload>),
56    /// Received transactions _from_ the peer
57    ReceivedTransaction(Transactions<N::BroadcastedTransaction>),
58    /// Broadcast transactions _from_ local _to_ a peer.
59    SendTransactions(SharedTransactions<N::BroadcastedTransaction>),
60    /// Broadcast cached pool transactions _from_ local _to_ a peer.
61    SendBroadcastPoolTransactions(BroadcastPoolTransactions),
62    /// Send new pooled transactions
63    PooledTransactions(NewPooledTransactionHashes),
64    /// All `eth` request variants.
65    EthRequest(PeerRequest<N>),
66    /// Announces when `BlockRange` is updated.
67    BlockRangeUpdated(BlockRangeUpdate),
68    /// Any other or manually crafted eth message.
69    ///
70    /// Caution: It is expected that this is a valid `eth_` capability message.
71    Other(RawCapabilityMessage),
72}
73
74impl<N: NetworkPrimitives> PeerMessage<N> {
75    /// Returns a static string identifying the message variant for logging.
76    pub const fn message_kind(&self) -> &'static str {
77        match self {
78            Self::NewBlockHashes(_) => "NewBlockHashes",
79            Self::NewBlock(_) => "NewBlock",
80            Self::ReceivedTransaction(_) => "ReceivedTransaction",
81            Self::SendTransactions(_) => "SendTransactions",
82            Self::SendBroadcastPoolTransactions(_) => "SendBroadcastPoolTransactions",
83            Self::PooledTransactions(_) => "PooledTransactions",
84            Self::EthRequest(_) => "EthRequest",
85            Self::BlockRangeUpdated(_) => "BlockRangeUpdated",
86            Self::Other(_) => "Other",
87        }
88    }
89
90    /// Returns `true` if this message is a broadcast (block/transaction announcement or
91    /// propagation) rather than a request/response.
92    pub const fn is_broadcast(&self) -> bool {
93        matches!(
94            self,
95            Self::NewBlockHashes(_) |
96                Self::NewBlock(_) |
97                Self::SendTransactions(_) |
98                Self::SendBroadcastPoolTransactions(_) |
99                Self::PooledTransactions(_)
100        )
101    }
102
103    /// Returns the number of items in the message payload, if applicable.
104    pub fn message_item_count(&self) -> usize {
105        match self {
106            Self::NewBlockHashes(msg) => msg.len(),
107            Self::ReceivedTransaction(msg) => msg.len(),
108            Self::SendTransactions(msg) => msg.len(),
109            Self::SendBroadcastPoolTransactions(msg) => msg.len(),
110            Self::PooledTransactions(msg) => msg.len(),
111            Self::NewBlock(_) |
112            Self::EthRequest(_) |
113            Self::BlockRangeUpdated(_) |
114            Self::Other(_) => 1,
115        }
116    }
117}
118
119/// Request Variants that only target block related data.
120#[derive(Debug, Clone, PartialEq, Eq)]
121pub enum BlockRequest {
122    /// Requests block headers from the peer.
123    ///
124    /// The response should be sent through the channel.
125    GetBlockHeaders(GetBlockHeaders),
126
127    /// Requests block bodies from the peer.
128    ///
129    /// The response should be sent through the channel.
130    GetBlockBodies(GetBlockBodies),
131    /// Requests block access lists from the peer.
132    ///
133    /// The response should be sent through the channel.
134    GetBlockAccessLists(GetBlockAccessLists),
135
136    /// Requests receipts from the peer.
137    ///
138    /// The response should be sent through the channel.
139    GetReceipts(GetReceipts),
140    /// Requests a `snap/2` (EIP-8189) message from the peer.
141    ///
142    /// The response should be sent through the channel. Boxed since `SnapProtocolMessage` is
143    /// large relative to the other variants.
144    GetSnap(Box<SnapProtocolMessage>),
145}
146
147/// Corresponding variant for [`PeerRequest`].
148#[derive(Debug)]
149pub enum PeerResponse<N: NetworkPrimitives = EthNetworkPrimitives> {
150    /// Represents a response to a request for block headers.
151    BlockHeaders {
152        /// The receiver channel for the response to a block headers request.
153        response: oneshot::Receiver<RequestResult<BlockHeaders<N::BlockHeader>>>,
154    },
155    /// Represents a response to a request for block bodies.
156    BlockBodies {
157        /// The receiver channel for the response to a block bodies request.
158        response: oneshot::Receiver<RequestResult<BlockBodies<N::BlockBody>>>,
159    },
160    /// Represents a response to a request for pooled transactions.
161    PooledTransactions {
162        /// The receiver channel for the response to a pooled transactions request.
163        response: oneshot::Receiver<RequestResult<PooledTransactions<N::PooledTransaction>>>,
164    },
165    /// Represents a response to a request for `NodeData`.
166    NodeData {
167        /// The receiver channel for the response to a `NodeData` request.
168        response: oneshot::Receiver<RequestResult<NodeData>>,
169    },
170    /// Represents a response to a request for receipts.
171    Receipts {
172        /// The receiver channel for the response to a receipts request.
173        response: oneshot::Receiver<RequestResult<Receipts<N::Receipt>>>,
174    },
175    /// Represents a response to a request for receipts.
176    ///
177    /// This is a variant of `Receipts` that was introduced in `eth/69`.
178    /// The difference is that this variant does not require the inclusion of bloom filters in the
179    /// response, making it more lightweight.
180    Receipts69 {
181        /// The receiver channel for the response to a receipts request.
182        response: oneshot::Receiver<RequestResult<Receipts69<N::Receipt>>>,
183    },
184    /// Represents a response to a request for receipts using eth/70.
185    Receipts70 {
186        /// The receiver channel for the response to a receipts request.
187        response: oneshot::Receiver<RequestResult<Receipts70<N::Receipt>>>,
188    },
189    /// Represents a response to a request for block access lists.
190    BlockAccessLists {
191        /// The receiver channel for the response to a block access lists request.
192        response: oneshot::Receiver<RequestResult<BlockAccessLists>>,
193    },
194    ///
195    /// Represents a response to a request for cells.
196    Cells {
197        /// The receiver channel for the response to a cells request.
198        response: oneshot::Receiver<RequestResult<Cells>>,
199    },
200    /// Represents a response to a `snap/2` (EIP-8189) request.
201    Snap {
202        /// The receiver channel for the response to a `snap/2` request.
203        response: oneshot::Receiver<RequestResult<SnapResponse>>,
204    },
205}
206
207// === impl PeerResponse ===
208
209impl<N: NetworkPrimitives> PeerResponse<N> {
210    /// Polls the type to completion.
211    pub(crate) fn poll(&mut self, cx: &mut Context<'_>) -> Poll<PeerResponseResult<N>> {
212        macro_rules! poll_request {
213            ($response:ident, $item:ident, $cx:ident) => {
214                match ready!($response.poll_unpin($cx)) {
215                    Ok(res) => PeerResponseResult::$item(res.map(|item| item.0)),
216                    Err(err) => PeerResponseResult::$item(Err(err.into())),
217                }
218            };
219        }
220
221        let res = match self {
222            Self::BlockHeaders { response } => {
223                poll_request!(response, BlockHeaders, cx)
224            }
225            Self::BlockBodies { response } => {
226                poll_request!(response, BlockBodies, cx)
227            }
228            Self::PooledTransactions { response } => {
229                poll_request!(response, PooledTransactions, cx)
230            }
231            Self::NodeData { response } => {
232                poll_request!(response, NodeData, cx)
233            }
234            Self::Receipts { response } => {
235                poll_request!(response, Receipts, cx)
236            }
237            Self::Receipts69 { response } => {
238                poll_request!(response, Receipts69, cx)
239            }
240            Self::Receipts70 { response } => match ready!(response.poll_unpin(cx)) {
241                Ok(res) => PeerResponseResult::Receipts70(res),
242                Err(err) => PeerResponseResult::Receipts70(Err(err.into())),
243            },
244            Self::BlockAccessLists { response } => match ready!(response.poll_unpin(cx)) {
245                Ok(res) => PeerResponseResult::BlockAccessLists(res),
246                Err(err) => PeerResponseResult::BlockAccessLists(Err(err.into())),
247            },
248            Self::Cells { response } => match ready!(response.poll_unpin(cx)) {
249                Ok(res) => PeerResponseResult::Cells(res),
250                Err(err) => PeerResponseResult::Cells(Err(err.into())),
251            },
252            Self::Snap { response } => match ready!(response.poll_unpin(cx)) {
253                Ok(res) => PeerResponseResult::Snap(res),
254                Err(err) => PeerResponseResult::Snap(Err(err.into())),
255            },
256        };
257        Poll::Ready(res)
258    }
259}
260
261/// All response variants for [`PeerResponse`]
262#[derive(Debug)]
263pub enum PeerResponseResult<N: NetworkPrimitives = EthNetworkPrimitives> {
264    /// Represents a result containing block headers or an error.
265    BlockHeaders(RequestResult<Vec<N::BlockHeader>>),
266    /// Represents a result containing block bodies or an error.
267    BlockBodies(RequestResult<Vec<N::BlockBody>>),
268    /// Represents a result containing pooled transactions or an error.
269    PooledTransactions(RequestResult<Vec<N::PooledTransaction>>),
270    /// Represents a result containing node data or an error.
271    NodeData(RequestResult<Vec<Bytes>>),
272    /// Represents a result containing receipts or an error.
273    Receipts(RequestResult<Vec<Vec<ReceiptWithBloom<N::Receipt>>>>),
274    /// Represents a result containing receipts or an error for eth/69.
275    Receipts69(RequestResult<Vec<Vec<N::Receipt>>>),
276    /// Represents a result containing receipts or an error for eth/70.
277    Receipts70(RequestResult<Receipts70<N::Receipt>>),
278    /// Represents a result containing block access lists or an error.
279    BlockAccessLists(RequestResult<BlockAccessLists>),
280    /// Represents a result containing cells or an error.
281    Cells(RequestResult<Cells>),
282    /// Represents a result containing a `snap/2` response or an error.
283    Snap(RequestResult<SnapResponse>),
284}
285
286// === impl PeerResponseResult ===
287
288impl<N: NetworkPrimitives> PeerResponseResult<N> {
289    /// Converts this response into the [`RequestMessage`] to send back to the peer: an
290    /// [`EthMessage`] for every variant except [`Self::Snap`], which becomes a
291    /// [`SnapProtocolMessage`].
292    pub fn try_into_message(self, id: u64) -> RequestResult<RequestMessage<N>> {
293        macro_rules! to_message {
294            ($response:ident, $item:ident, $request_id:ident) => {
295                match $response {
296                    Ok(res) => {
297                        let request = RequestPair { request_id: $request_id, message: $item(res) };
298                        Ok(RequestMessage::Eth(EthMessage::$item(request)))
299                    }
300                    Err(err) => Err(err),
301                }
302            };
303        }
304        match self {
305            Self::BlockHeaders(resp) => {
306                to_message!(resp, BlockHeaders, id)
307            }
308            Self::BlockBodies(resp) => {
309                to_message!(resp, BlockBodies, id)
310            }
311            Self::PooledTransactions(resp) => {
312                to_message!(resp, PooledTransactions, id)
313            }
314            Self::NodeData(resp) => {
315                to_message!(resp, NodeData, id)
316            }
317            Self::Receipts(resp) => {
318                to_message!(resp, Receipts, id)
319            }
320            Self::Receipts69(resp) => {
321                to_message!(resp, Receipts69, id)
322            }
323            Self::Receipts70(resp) => match resp {
324                Ok(res) => {
325                    let request = RequestPair { request_id: id, message: res };
326                    Ok(RequestMessage::Eth(EthMessage::Receipts70(request)))
327                }
328                Err(err) => Err(err),
329            },
330            Self::BlockAccessLists(resp) => match resp {
331                Ok(res) => {
332                    let request = RequestPair { request_id: id, message: res };
333                    Ok(RequestMessage::Eth(EthMessage::BlockAccessLists(request)))
334                }
335                Err(err) => Err(err),
336            },
337            Self::Cells(resp) => match resp {
338                Ok(res) => {
339                    let request = RequestPair { request_id: id, message: res };
340                    Ok(RequestMessage::Eth(EthMessage::Cells(request)))
341                }
342                Err(err) => Err(err),
343            },
344            Self::Snap(resp) => match resp {
345                Ok(res) => {
346                    let mut message: SnapProtocolMessage = res.into();
347                    message.set_request_id(id);
348                    Ok(RequestMessage::Snap(message))
349                }
350                Err(err) => Err(err),
351            },
352        }
353    }
354
355    /// Returns the `Err` value if the result is an error.
356    pub fn err(&self) -> Option<&RequestError> {
357        match self {
358            Self::BlockHeaders(res) => res.as_ref().err(),
359            Self::BlockBodies(res) => res.as_ref().err(),
360            Self::PooledTransactions(res) => res.as_ref().err(),
361            Self::NodeData(res) => res.as_ref().err(),
362            Self::Receipts(res) => res.as_ref().err(),
363            Self::Receipts69(res) => res.as_ref().err(),
364            Self::Receipts70(res) => res.as_ref().err(),
365            Self::BlockAccessLists(res) => res.as_ref().err(),
366            Self::Cells(res) => res.as_ref().err(),
367            Self::Snap(res) => res.as_ref().err(),
368        }
369    }
370
371    /// Returns whether this result is an error.
372    pub fn is_err(&self) -> bool {
373        self.err().is_some()
374    }
375}
reth_network/message.rs

reth_network/
message.rs
