reth_network_api/
events.rs

1//! API related to listening for network events.
2
3use reth_eth_wire_types::{
4    message::RequestPair, BlockAccessLists, BlockBodies, BlockHeaders, Capabilities, Cells,
5    DisconnectReason, EthMessage, EthNetworkPrimitives, EthVersion, GetBlockAccessLists,
6    GetBlockBodies, GetBlockHeaders, GetCells, GetNodeData, GetPooledTransactions, GetReceipts,
7    GetReceipts70, NetworkPrimitives, NodeData, PooledTransactions, Receipts, Receipts69,
8    Receipts70, UnifiedStatus,
9};
10use reth_ethereum_forks::ForkId;
11use reth_network_p2p::error::{RequestError, RequestResult};
12use reth_network_peers::{NodeRecord, PeerId};
13use reth_network_types::{PeerAddr, PeerKind};
14use reth_tokio_util::EventStream;
15use std::{
16    fmt,
17    net::SocketAddr,
18    pin::Pin,
19    sync::Arc,
20    task::{Context, Poll},
21};
22use tokio::sync::{mpsc, oneshot};
23use tokio_stream::{wrappers::UnboundedReceiverStream, Stream, StreamExt};
24
25/// A boxed stream of network peer events that provides a type-erased interface.
26pub struct PeerEventStream(Pin<Box<dyn Stream<Item = PeerEvent> + Send + Sync>>);
27
28impl fmt::Debug for PeerEventStream {
29    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30        f.debug_struct("PeerEventStream").finish_non_exhaustive()
31    }
32}
33
34impl PeerEventStream {
35    /// Create a new stream [`PeerEventStream`] by converting the provided stream's items into peer
36    /// events [`PeerEvent`]
37    pub fn new<S, T>(stream: S) -> Self
38    where
39        S: Stream<Item = T> + Send + Sync + 'static,
40        T: Into<PeerEvent> + 'static,
41    {
42        let mapped_stream = stream.map(Into::into);
43        Self(Box::pin(mapped_stream))
44    }
45}
46
47impl Stream for PeerEventStream {
48    type Item = PeerEvent;
49
50    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
51        self.0.as_mut().poll_next(cx)
52    }
53}
54
55/// Represents information about an established peer session.
56#[derive(Debug, Clone)]
57pub struct SessionInfo {
58    /// The identifier of the peer to which a session was established.
59    pub peer_id: PeerId,
60    /// The remote addr of the peer to which a session was established.
61    pub remote_addr: SocketAddr,
62    /// The client version of the peer to which a session was established.
63    pub client_version: Arc<str>,
64    /// Capabilities the peer announced.
65    pub capabilities: Arc<Capabilities>,
66    /// The status of the peer to which a session was established.
67    pub status: Arc<UnifiedStatus>,
68    /// Negotiated eth version of the session.
69    pub version: EthVersion,
70    /// The kind of peer this session represents
71    pub peer_kind: PeerKind,
72}
73
74/// (Non-exhaustive) List of the different events emitted by the network that are of interest for
75/// subscribers.
76///
77/// This includes any event types that may be relevant to tasks, for metrics, keep track of peers
78/// etc.
79#[derive(Debug, Clone)]
80pub enum PeerEvent {
81    /// Closed the peer session.
82    SessionClosed {
83        /// The identifier of the peer to which a session was closed.
84        peer_id: PeerId,
85        /// Why the disconnect was triggered
86        reason: Option<DisconnectReason>,
87    },
88    /// Established a new session with the given peer.
89    SessionEstablished(SessionInfo),
90    /// Event emitted when a new peer is added
91    PeerAdded(PeerId),
92    /// Event emitted when a new peer is removed
93    PeerRemoved(PeerId),
94}
95
96/// (Non-exhaustive) Network events representing peer lifecycle events and session requests.
97#[derive(Debug)]
98pub enum NetworkEvent<R = PeerRequest> {
99    /// Basic peer lifecycle event.
100    Peer(PeerEvent),
101    /// Session established with requests.
102    ActivePeerSession {
103        /// Session information
104        info: SessionInfo,
105        /// A request channel to the session task.
106        messages: PeerRequestSender<R>,
107    },
108}
109
110impl<R> Clone for NetworkEvent<R> {
111    fn clone(&self) -> Self {
112        match self {
113            Self::Peer(event) => Self::Peer(event.clone()),
114            Self::ActivePeerSession { info, messages } => {
115                Self::ActivePeerSession { info: info.clone(), messages: messages.clone() }
116            }
117        }
118    }
119}
120
121impl<R> From<NetworkEvent<R>> for PeerEvent {
122    fn from(event: NetworkEvent<R>) -> Self {
123        match event {
124            NetworkEvent::Peer(peer_event) => peer_event,
125            NetworkEvent::ActivePeerSession { info, .. } => Self::SessionEstablished(info),
126        }
127    }
128}
129
130/// Provides peer event subscription for the network.
131#[auto_impl::auto_impl(&, Arc)]
132pub trait NetworkPeersEvents: Send + Sync {
133    /// Creates a new peer event listener stream.
134    fn peer_events(&self) -> PeerEventStream;
135}
136
137/// Provides event subscription for the network.
138#[auto_impl::auto_impl(&, Arc)]
139pub trait NetworkEventListenerProvider: NetworkPeersEvents {
140    /// The primitive types to use in the `PeerRequest` used in the stream.
141    type Primitives: NetworkPrimitives;
142
143    /// Creates a new [`NetworkEvent`] listener channel.
144    fn event_listener(&self) -> EventStream<NetworkEvent<PeerRequest<Self::Primitives>>>;
145    /// Returns a new [`DiscoveryEvent`] stream.
146    ///
147    /// This stream yields [`DiscoveryEvent`]s for each peer that is discovered.
148    fn discovery_listener(&self) -> UnboundedReceiverStream<DiscoveryEvent>;
149}
150
151/// Events produced by the `Discovery` manager.
152#[derive(Debug, Clone, PartialEq, Eq)]
153pub enum DiscoveryEvent {
154    /// Discovered a node
155    NewNode(DiscoveredEvent),
156    /// Retrieved a [`ForkId`] from the peer via ENR request.
157    ///
158    /// Contains the full [`NodeRecord`] (peer ID + address) and the reported [`ForkId`].
159    /// Used to verify fork compatibility before admitting the peer.
160    ///
161    /// See also <https://eips.ethereum.org/EIPS/eip-868>
162    EnrForkId(NodeRecord, ForkId),
163}
164
165/// Represents events related to peer discovery in the network.
166#[derive(Debug, Clone, PartialEq, Eq)]
167pub enum DiscoveredEvent {
168    /// Indicates that a new peer has been discovered and queued for potential connection.
169    ///
170    /// This event is generated when the system becomes aware of a new peer
171    /// but hasn't yet established a connection.
172    ///
173    /// # Fields
174    ///
175    /// * `peer_id` - The unique identifier of the discovered peer.
176    /// * `addr` - The network address of the discovered peer.
177    /// * `fork_id` - An optional identifier for the fork that this peer is associated with. `None`
178    ///   if the peer is not associated with a specific fork.
179    EventQueued {
180        /// The unique identifier of the discovered peer.
181        peer_id: PeerId,
182        /// The network address of the discovered peer.
183        addr: PeerAddr,
184        /// An optional identifier for the fork that this peer is associated with.
185        /// `None` if the peer is not associated with a specific fork.
186        fork_id: Option<ForkId>,
187    },
188}
189
190/// Protocol related request messages that expect a response
191#[derive(Debug)]
192pub enum PeerRequest<N: NetworkPrimitives = EthNetworkPrimitives> {
193    /// Requests block headers from the peer.
194    ///
195    /// The response should be sent through the channel.
196    GetBlockHeaders {
197        /// The request for block headers.
198        request: GetBlockHeaders,
199        /// The channel to send the response for block headers.
200        response: oneshot::Sender<RequestResult<BlockHeaders<N::BlockHeader>>>,
201    },
202    /// Requests block bodies from the peer.
203    ///
204    /// The response should be sent through the channel.
205    GetBlockBodies {
206        /// The request for block bodies.
207        request: GetBlockBodies,
208        /// The channel to send the response for block bodies.
209        response: oneshot::Sender<RequestResult<BlockBodies<N::BlockBody>>>,
210    },
211    /// Requests pooled transactions from the peer.
212    ///
213    /// The response should be sent through the channel.
214    GetPooledTransactions {
215        /// The request for pooled transactions.
216        request: GetPooledTransactions,
217        /// The channel to send the response for pooled transactions.
218        response: oneshot::Sender<RequestResult<PooledTransactions<N::PooledTransaction>>>,
219    },
220    /// Requests `NodeData` from the peer.
221    ///
222    /// The response should be sent through the channel.
223    GetNodeData {
224        /// The request for `NodeData`.
225        request: GetNodeData,
226        /// The channel to send the response for `NodeData`.
227        response: oneshot::Sender<RequestResult<NodeData>>,
228    },
229    /// Requests receipts from the peer.
230    ///
231    /// The response should be sent through the channel.
232    GetReceipts {
233        /// The request for receipts.
234        request: GetReceipts,
235        /// The channel to send the response for receipts.
236        response: oneshot::Sender<RequestResult<Receipts<N::Receipt>>>,
237    },
238    /// Requests receipts from the peer without bloom filter.
239    ///
240    /// The response should be sent through the channel.
241    GetReceipts69 {
242        /// The request for receipts.
243        request: GetReceipts,
244        /// The channel to send the response for receipts.
245        response: oneshot::Sender<RequestResult<Receipts69<N::Receipt>>>,
246    },
247    /// Requests receipts from the peer using eth/70 (supports `firstBlockReceiptIndex`).
248    ///
249    /// The response should be sent through the channel.
250    GetReceipts70 {
251        /// The request for receipts.
252        request: GetReceipts70,
253        /// The channel to send the response for receipts.
254        response: oneshot::Sender<RequestResult<Receipts70<N::Receipt>>>,
255    },
256    /// Requests block access lists from the peer.
257    ///
258    /// The response should be sent through the channel.
259    GetBlockAccessLists {
260        /// The request for block access lists.
261        request: GetBlockAccessLists,
262        /// The channel to send the response for block access lists.
263        response: oneshot::Sender<RequestResult<BlockAccessLists>>,
264    },
265    /// Requests cells from the peer.
266    ///
267    /// The response should be sent through the channel.
268    GetCells {
269        /// The request for cells.
270        request: GetCells,
271        /// The channel to send the response for cells.
272        response: oneshot::Sender<RequestResult<Cells>>,
273    },
274}
275
276// === impl PeerRequest ===
277
278impl<N: NetworkPrimitives> PeerRequest<N> {
279    /// Invoked if we received a response which does not match the request
280    pub fn send_bad_response(self) {
281        self.send_err_response(RequestError::BadResponse)
282    }
283
284    /// Send an error back to the receiver.
285    pub fn send_err_response(self, err: RequestError) {
286        let _ = match self {
287            Self::GetBlockHeaders { response, .. } => response.send(Err(err)).ok(),
288            Self::GetBlockBodies { response, .. } => response.send(Err(err)).ok(),
289            Self::GetPooledTransactions { response, .. } => response.send(Err(err)).ok(),
290            Self::GetNodeData { response, .. } => response.send(Err(err)).ok(),
291            Self::GetReceipts { response, .. } => response.send(Err(err)).ok(),
292            Self::GetReceipts69 { response, .. } => response.send(Err(err)).ok(),
293            Self::GetReceipts70 { response, .. } => response.send(Err(err)).ok(),
294            Self::GetBlockAccessLists { response, .. } => response.send(Err(err)).ok(),
295            Self::GetCells { response, .. } => response.send(Err(err)).ok(),
296        };
297    }
298
299    /// Returns true if this request is supported for the negotiated eth protocol version.
300    #[inline]
301    pub fn is_supported_by_eth_version(&self, version: EthVersion) -> bool {
302        match self {
303            Self::GetBlockAccessLists { .. } => version >= EthVersion::Eth71,
304            Self::GetCells { .. } => version >= EthVersion::Eth72,
305            _ => true,
306        }
307    }
308
309    /// Returns the [`EthMessage`] for this type
310    pub fn create_request_message(&self, request_id: u64) -> EthMessage<N> {
311        match self {
312            Self::GetBlockHeaders { request, .. } => {
313                EthMessage::GetBlockHeaders(RequestPair { request_id, message: *request })
314            }
315            Self::GetBlockBodies { request, .. } => {
316                EthMessage::GetBlockBodies(RequestPair { request_id, message: request.clone() })
317            }
318            Self::GetPooledTransactions { request, .. } => {
319                EthMessage::GetPooledTransactions(RequestPair {
320                    request_id,
321                    message: request.clone(),
322                })
323            }
324            Self::GetNodeData { request, .. } => {
325                EthMessage::GetNodeData(RequestPair { request_id, message: request.clone() })
326            }
327            Self::GetReceipts { request, .. } | Self::GetReceipts69 { request, .. } => {
328                EthMessage::GetReceipts(RequestPair { request_id, message: request.clone() })
329            }
330            Self::GetReceipts70 { request, .. } => {
331                EthMessage::GetReceipts70(RequestPair { request_id, message: request.clone() })
332            }
333            Self::GetBlockAccessLists { request, .. } => {
334                EthMessage::GetBlockAccessLists(RequestPair {
335                    request_id,
336                    message: request.clone(),
337                })
338            }
339            Self::GetCells { request, .. } => {
340                EthMessage::GetCells(RequestPair { request_id, message: request.clone() })
341            }
342        }
343    }
344
345    /// Consumes the type and returns the inner [`GetPooledTransactions`] variant.
346    pub fn into_get_pooled_transactions(self) -> Option<GetPooledTransactions> {
347        match self {
348            Self::GetPooledTransactions { request, .. } => Some(request),
349            _ => None,
350        }
351    }
352}
353
354/// A Cloneable connection for sending _requests_ directly to the session of a peer.
355pub struct PeerRequestSender<R = PeerRequest> {
356    /// id of the remote node.
357    pub peer_id: PeerId,
358    /// The Sender half connected to a session.
359    pub to_session_tx: mpsc::Sender<R>,
360}
361
362impl<R> Clone for PeerRequestSender<R> {
363    fn clone(&self) -> Self {
364        Self { peer_id: self.peer_id, to_session_tx: self.to_session_tx.clone() }
365    }
366}
367
368// === impl PeerRequestSender ===
369
370impl<R> PeerRequestSender<R> {
371    /// Constructs a new sender instance that's wired to a session
372    pub const fn new(peer_id: PeerId, to_session_tx: mpsc::Sender<R>) -> Self {
373        Self { peer_id, to_session_tx }
374    }
375
376    /// Attempts to immediately send a message on this Sender
377    pub fn try_send(&self, req: R) -> Result<(), mpsc::error::TrySendError<R>> {
378        self.to_session_tx.try_send(req)
379    }
380
381    /// Returns the peer id of the remote peer.
382    pub const fn peer_id(&self) -> &PeerId {
383        &self.peer_id
384    }
385}
386
387impl<R> fmt::Debug for PeerRequestSender<R> {
388    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
389        f.debug_struct("PeerRequestSender").field("peer_id", &self.peer_id).finish_non_exhaustive()
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use super::*;
396
397    #[test]
398    fn test_get_block_access_lists_version_support() {
399        let (tx, _rx) = oneshot::channel();
400        let req: PeerRequest<EthNetworkPrimitives> =
401            PeerRequest::GetBlockAccessLists { request: GetBlockAccessLists(vec![]), response: tx };
402
403        assert!(!req.is_supported_by_eth_version(EthVersion::Eth70));
404        assert!(req.is_supported_by_eth_version(EthVersion::Eth71));
405    }
406
407    #[test]
408    fn test_get_cells_version_support() {
409        let (tx, _rx) = oneshot::channel();
410        let req: PeerRequest<EthNetworkPrimitives> =
411            PeerRequest::GetCells { request: GetCells::default(), response: tx };
412
413        assert!(!req.is_supported_by_eth_version(EthVersion::Eth71));
414        assert!(req.is_supported_by_eth_version(EthVersion::Eth72));
415    }
416}
reth_network_api/events.rs

reth_network_api/
events.rs
