reth_network_api/
events.rs

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

reth_network_api/
events.rs
