reth_network/
manager.rs

1//! High level network management.
2//!
3//! The [`NetworkManager`] contains the state of the network as a whole. It controls how connections
4//! are handled and keeps track of connections to peers.
5//!
6//! ## Capabilities
7//!
8//! The network manages peers depending on their announced capabilities via their `RLPx` sessions. Most importantly the [Ethereum Wire Protocol](https://github.com/ethereum/devp2p/blob/master/caps/eth.md)(`eth`).
9//!
10//! ## Overview
11//!
12//! The [`NetworkManager`] is responsible for advancing the state of the `network`. The `network` is
13//! made up of peer-to-peer connections between nodes that are available on the same network.
14//! Responsible for peer discovery is ethereum's discovery protocol (discv4, discv5). If the address
15//! (IP+port) of our node is published via discovery, remote peers can initiate inbound connections
16//! to the local node. Once a (tcp) connection is established, both peers start to authenticate a [RLPx session](https://github.com/ethereum/devp2p/blob/master/rlpx.md) via a handshake. If the handshake was successful, both peers announce their capabilities and are now ready to exchange sub-protocol messages via the `RLPx` session.
17
18use crate::{
19    budget::{DEFAULT_BUDGET_TRY_DRAIN_NETWORK_HANDLE_CHANNEL, DEFAULT_BUDGET_TRY_DRAIN_SWARM},
20    config::NetworkConfig,
21    discovery::Discovery,
22    error::{NetworkError, ServiceKind},
23    eth_requests::IncomingEthRequest,
24    import::{BlockImport, BlockImportEvent, BlockImportOutcome, BlockValidation, NewBlockEvent},
25    listener::ConnectionListener,
26    message::{NewBlockMessage, PeerMessage},
27    metrics::{
28        BackedOffPeersMetrics, ClosedSessionsMetrics, DirectionalDisconnectMetrics, NetworkMetrics,
29        PendingSessionFailureMetrics,
30    },
31    network::{NetworkHandle, NetworkHandleMessage},
32    peers::{BackoffReason, PeersManager},
33    poll_nested_stream_with_budget,
34    protocol::IntoRlpxSubProtocol,
35    required_block_filter::RequiredBlockFilter,
36    session::SessionManager,
37    state::NetworkState,
38    swarm::{Swarm, SwarmEvent},
39    transactions::NetworkTransactionEvent,
40    FetchClient, NetworkBuilder,
41};
42use futures::{Future, StreamExt};
43use parking_lot::Mutex;
44use reth_chainspec::EnrForkIdEntry;
45use reth_eth_wire::{DisconnectReason, EthNetworkPrimitives, NetworkPrimitives};
46use reth_fs_util::{self as fs, FsPathError};
47use reth_metrics::common::mpsc::MemoryBoundedSender;
48use reth_network_api::{
49    events::{PeerEvent, SessionInfo},
50    test_utils::PeersHandle,
51    EthProtocolInfo, NetworkEvent, NetworkStatus, PeerInfo, PeerRequest,
52};
53use reth_network_peers::{NodeRecord, PeerId};
54use reth_network_types::ReputationChangeKind;
55use reth_storage_api::BlockNumReader;
56use reth_tasks::shutdown::GracefulShutdown;
57use reth_tokio_util::EventSender;
58use secp256k1::SecretKey;
59use std::{
60    net::SocketAddr,
61    path::Path,
62    pin::Pin,
63    sync::{
64        atomic::{AtomicU64, AtomicUsize, Ordering},
65        Arc,
66    },
67    task::{Context, Poll},
68    time::{Duration, Instant},
69};
70use tokio::sync::mpsc::{self, error::TrySendError};
71use tokio_stream::wrappers::UnboundedReceiverStream;
72use tracing::{debug, error, trace, warn};
73
74#[cfg_attr(doc, aquamarine::aquamarine)]
75// TODO: Inlined diagram due to a bug in aquamarine library, should become an include when it's
76// fixed. See https://github.com/mersinvald/aquamarine/issues/50
77// include_mmd!("docs/mermaid/network-manager.mmd")
78/// Manages the _entire_ state of the network.
79///
80/// This is an endless [`Future`] that consistently drives the state of the entire network forward.
81///
82/// The [`NetworkManager`] is the container type for all parts involved with advancing the network.
83///
84/// ```mermaid
85/// graph TB
86///   handle(NetworkHandle)
87///   events(NetworkEvents)
88///   transactions(Transactions Task)
89///   ethrequest(ETH Request Task)
90///   discovery(Discovery Task)
91///   subgraph NetworkManager
92///     direction LR
93///     subgraph Swarm
94///         direction TB
95///         B1[(Session Manager)]
96///         B2[(Connection Listener)]
97///         B3[(Network State)]
98///     end
99///  end
100///  handle <--> |request response channel| NetworkManager
101///  NetworkManager --> |Network events| events
102///  transactions <--> |transactions| NetworkManager
103///  ethrequest <--> |ETH request handing| NetworkManager
104///  discovery --> |Discovered peers| NetworkManager
105/// ```
106#[derive(Debug)]
107#[must_use = "The NetworkManager does nothing unless polled"]
108pub struct NetworkManager<N: NetworkPrimitives = EthNetworkPrimitives> {
109    /// The type that manages the actual network part, which includes connections.
110    swarm: Swarm<N>,
111    /// Underlying network handle that can be shared.
112    handle: NetworkHandle<N>,
113    /// Receiver half of the command channel set up between this type and the [`NetworkHandle`]
114    from_handle_rx: UnboundedReceiverStream<NetworkHandleMessage<N>>,
115    /// Handles block imports according to the `eth` protocol.
116    block_import: Box<dyn BlockImport<N::NewBlockPayload>>,
117    /// Sender for high level network events.
118    event_sender: EventSender<NetworkEvent<PeerRequest<N>>>,
119    /// Sender half to send events to the
120    /// [`TransactionsManager`](crate::transactions::TransactionsManager) task, if configured.
121    to_transactions_manager: Option<MemoryBoundedSender<NetworkTransactionEvent<N>>>,
122    /// Sender half to send events to the
123    /// [`EthRequestHandler`](crate::eth_requests::EthRequestHandler) task, if configured.
124    ///
125    /// The channel that originally receives and bundles all requests from all sessions is already
126    /// bounded. However, since handling an eth request is more I/O intensive than delegating
127    /// them from the bounded channel to the eth-request channel, it is possible that this
128    /// builds up if the node is flooded with requests.
129    ///
130    /// Even though nonmalicious requests are relatively cheap, it's possible to craft
131    /// body requests with bogus data up until the allowed max message size limit.
132    /// Thus, we use a bounded channel here to avoid unbounded build up if the node is flooded with
133    /// requests. This channel size is set at
134    /// [`ETH_REQUEST_CHANNEL_CAPACITY`](crate::builder::ETH_REQUEST_CHANNEL_CAPACITY)
135    to_eth_request_handler: Option<mpsc::Sender<IncomingEthRequest<N>>>,
136    /// Tracks the number of active session (connected peers).
137    ///
138    /// This is updated via internal events and shared via `Arc` with the [`NetworkHandle`]
139    /// Updated by the `NetworkWorker` and loaded by the `NetworkService`.
140    num_active_peers: Arc<AtomicUsize>,
141    /// Metrics for the Network
142    metrics: NetworkMetrics,
143    /// Disconnect metrics for the Network, split by connection direction.
144    disconnect_metrics: DirectionalDisconnectMetrics,
145    /// Closed sessions metrics, split by direction.
146    closed_sessions_metrics: ClosedSessionsMetrics,
147    /// Pending session failure metrics, split by direction.
148    pending_session_failure_metrics: PendingSessionFailureMetrics,
149    /// Backed off peers metrics, split by reason.
150    backed_off_peers_metrics: BackedOffPeersMetrics,
151}
152
153impl NetworkManager {
154    /// Creates the manager of a new network with [`EthNetworkPrimitives`] types.
155    ///
156    /// ```no_run
157    /// # async fn f() {
158    /// use reth_chainspec::MAINNET;
159    /// use reth_network::{NetworkConfig, NetworkManager};
160    /// use reth_tasks::Runtime;
161    /// let config = NetworkConfig::builder_with_rng_secret_key(Runtime::test())
162    ///     .build_with_noop_provider(MAINNET.clone());
163    /// let manager = NetworkManager::eth(config).await;
164    /// # }
165    /// ```
166    pub async fn eth<C: BlockNumReader + 'static>(
167        config: NetworkConfig<C, EthNetworkPrimitives>,
168    ) -> Result<Self, NetworkError> {
169        Self::new(config).await
170    }
171}
172
173impl<N: NetworkPrimitives> NetworkManager<N> {
174    /// Sets the dedicated channel for events intended for the
175    /// [`TransactionsManager`](crate::transactions::TransactionsManager).
176    pub fn with_transactions(
177        mut self,
178        tx: MemoryBoundedSender<NetworkTransactionEvent<N>>,
179    ) -> Self {
180        self.set_transactions(tx);
181        self
182    }
183
184    /// Sets the dedicated channel for events intended for the
185    /// [`TransactionsManager`](crate::transactions::TransactionsManager).
186    pub fn set_transactions(&mut self, tx: MemoryBoundedSender<NetworkTransactionEvent<N>>) {
187        self.to_transactions_manager = Some(tx);
188    }
189
190    /// Sets the dedicated channel for events intended for the
191    /// [`EthRequestHandler`](crate::eth_requests::EthRequestHandler).
192    pub fn with_eth_request_handler(mut self, tx: mpsc::Sender<IncomingEthRequest<N>>) -> Self {
193        self.set_eth_request_handler(tx);
194        self
195    }
196
197    /// Sets the dedicated channel for events intended for the
198    /// [`EthRequestHandler`](crate::eth_requests::EthRequestHandler).
199    pub fn set_eth_request_handler(&mut self, tx: mpsc::Sender<IncomingEthRequest<N>>) {
200        self.to_eth_request_handler = Some(tx);
201    }
202
203    /// Adds an additional protocol handler to the `RLPx` sub-protocol list.
204    pub fn add_rlpx_sub_protocol(&mut self, protocol: impl IntoRlpxSubProtocol) {
205        self.swarm.add_rlpx_sub_protocol(protocol)
206    }
207
208    /// Returns the [`NetworkHandle`] that can be cloned and shared.
209    ///
210    /// The [`NetworkHandle`] can be used to interact with this [`NetworkManager`]
211    pub const fn handle(&self) -> &NetworkHandle<N> {
212        &self.handle
213    }
214
215    /// Returns the secret key used for authenticating sessions.
216    pub const fn secret_key(&self) -> SecretKey {
217        self.swarm.sessions().secret_key()
218    }
219
220    #[inline]
221    fn update_poll_metrics(&self, start: Instant, poll_durations: NetworkManagerPollDurations) {
222        let metrics = &self.metrics;
223
224        let NetworkManagerPollDurations { acc_network_handle, acc_swarm } = poll_durations;
225
226        // update metrics for whole poll function
227        metrics.duration_poll_network_manager.set(start.elapsed().as_secs_f64());
228        // update poll metrics for nested items
229        metrics.acc_duration_poll_network_handle.set(acc_network_handle.as_secs_f64());
230        metrics.acc_duration_poll_swarm.set(acc_swarm.as_secs_f64());
231    }
232
233    /// Creates the manager of a new network.
234    ///
235    /// The [`NetworkManager`] is an endless future that needs to be polled in order to advance the
236    /// state of the entire network.
237    pub async fn new<C: BlockNumReader + 'static>(
238        config: NetworkConfig<C, N>,
239    ) -> Result<Self, NetworkError> {
240        let NetworkConfig {
241            client,
242            secret_key,
243            discovery_v4_addr,
244            mut discovery_v4_config,
245            mut discovery_v5_config,
246            listener_addr,
247            peers_config,
248            sessions_config,
249            chain_id,
250            block_import,
251            network_mode,
252            boot_nodes,
253            executor,
254            hello_message,
255            status,
256            fork_filter,
257            dns_discovery_config,
258            extra_protocols,
259            tx_gossip_disabled,
260            transactions_manager_config: _,
261            nat,
262            handshake,
263            eth_max_message_size,
264            required_block_hashes,
265        } = config;
266
267        let peers_manager = PeersManager::new(peers_config);
268        let peers_handle = peers_manager.handle();
269
270        let incoming = ConnectionListener::bind(listener_addr).await.map_err(|err| {
271            NetworkError::from_io_error(err, ServiceKind::Listener(listener_addr))
272        })?;
273
274        // retrieve the tcp address of the socket
275        let listener_addr = incoming.local_address();
276
277        // resolve boot nodes
278        let resolved_boot_nodes =
279            futures::future::try_join_all(boot_nodes.iter().map(|record| record.resolve())).await?;
280
281        if let Some(disc_config) = discovery_v4_config.as_mut() {
282            // merge configured boot nodes
283            disc_config.bootstrap_nodes.extend(resolved_boot_nodes.clone());
284            // add the forkid entry for EIP-868, but wrap it in an `EnrForkIdEntry` for proper
285            // encoding
286            disc_config.add_eip868_pair("eth", EnrForkIdEntry::from(status.forkid));
287        }
288
289        if let Some(discv5) = discovery_v5_config.as_mut() {
290            // merge configured boot nodes
291            discv5.extend_unsigned_boot_nodes(resolved_boot_nodes)
292        }
293
294        let discovery = Discovery::new(
295            listener_addr,
296            discovery_v4_addr,
297            secret_key,
298            discovery_v4_config,
299            discovery_v5_config,
300            dns_discovery_config,
301        )
302        .await?;
303        // need to retrieve the addr here since provided port could be `0`
304        let local_peer_id = discovery.local_id();
305        let discv4 = discovery.discv4();
306        let discv5 = discovery.discv5();
307
308        let num_active_peers = Arc::new(AtomicUsize::new(0));
309
310        let sessions = SessionManager::new(
311            secret_key,
312            sessions_config,
313            executor,
314            status,
315            hello_message,
316            fork_filter,
317            extra_protocols,
318            handshake,
319            eth_max_message_size,
320            network_mode.is_stake(),
321        );
322
323        let state = NetworkState::new(
324            crate::state::BlockNumReader::new(client),
325            discovery,
326            peers_manager,
327            Arc::clone(&num_active_peers),
328        );
329
330        let swarm = Swarm::new(incoming, sessions, state);
331
332        let (to_manager_tx, from_handle_rx) = mpsc::unbounded_channel();
333
334        let event_sender: EventSender<NetworkEvent<PeerRequest<N>>> = Default::default();
335
336        let handle = NetworkHandle::new(
337            Arc::clone(&num_active_peers),
338            Arc::new(Mutex::new(listener_addr)),
339            to_manager_tx,
340            secret_key,
341            local_peer_id,
342            peers_handle,
343            network_mode,
344            Arc::new(AtomicU64::new(chain_id)),
345            tx_gossip_disabled,
346            discv4,
347            discv5,
348            event_sender.clone(),
349            nat,
350        );
351
352        // Spawn required block peer filter if configured
353        if !required_block_hashes.is_empty() {
354            let filter = RequiredBlockFilter::new(handle.clone(), required_block_hashes);
355            filter.spawn();
356        }
357
358        Ok(Self {
359            swarm,
360            handle,
361            from_handle_rx: UnboundedReceiverStream::new(from_handle_rx),
362            block_import,
363            event_sender,
364            to_transactions_manager: None,
365            to_eth_request_handler: None,
366            num_active_peers,
367            metrics: Default::default(),
368            disconnect_metrics: Default::default(),
369            closed_sessions_metrics: Default::default(),
370            pending_session_failure_metrics: Default::default(),
371            backed_off_peers_metrics: Default::default(),
372        })
373    }
374
375    /// Create a new [`NetworkManager`] instance and start a [`NetworkBuilder`] to configure all
376    /// components of the network
377    ///
378    /// ```
379    /// use reth_network::{
380    ///     config::rng_secret_key, EthNetworkPrimitives, NetworkConfig, NetworkManager,
381    /// };
382    /// use reth_network_peers::mainnet_nodes;
383    /// use reth_storage_api::noop::NoopProvider;
384    /// use reth_tasks::Runtime;
385    /// use reth_transaction_pool::TransactionPool;
386    /// async fn launch<Pool: TransactionPool>(pool: Pool) {
387    ///     // This block provider implementation is used for testing purposes.
388    ///     let client = NoopProvider::default();
389    ///
390    ///     // The key that's used for encrypting sessions and to identify our node.
391    ///     let local_key = rng_secret_key();
392    ///
393    ///     let config = NetworkConfig::<_, EthNetworkPrimitives>::builder(local_key, Runtime::test())
394    ///         .boot_nodes(mainnet_nodes())
395    ///         .build(client.clone());
396    ///     let transactions_manager_config = config.transactions_manager_config.clone();
397    ///
398    ///     // create the network instance
399    ///     let (handle, network, transactions, request_handler) = NetworkManager::builder(config)
400    ///         .await
401    ///         .unwrap()
402    ///         .transactions(pool, transactions_manager_config)
403    ///         .request_handler(client)
404    ///         .split_with_handle();
405    /// }
406    /// ```
407    pub async fn builder<C: BlockNumReader + 'static>(
408        config: NetworkConfig<C, N>,
409    ) -> Result<NetworkBuilder<(), (), N>, NetworkError> {
410        let network = Self::new(config).await?;
411        Ok(network.into_builder())
412    }
413
414    /// Create a [`NetworkBuilder`] to configure all components of the network
415    pub const fn into_builder(self) -> NetworkBuilder<(), (), N> {
416        NetworkBuilder { network: self, transactions: (), request_handler: () }
417    }
418
419    /// Returns the [`SocketAddr`] that listens for incoming tcp connections.
420    pub const fn local_addr(&self) -> SocketAddr {
421        self.swarm.listener().local_address()
422    }
423
424    /// How many peers we're currently connected to.
425    pub fn num_connected_peers(&self) -> usize {
426        self.swarm.state().num_active_peers()
427    }
428
429    /// Returns the [`PeerId`] used in the network.
430    pub fn peer_id(&self) -> &PeerId {
431        self.handle.peer_id()
432    }
433
434    /// Returns an iterator over all peers in the peer set.
435    pub fn all_peers(&self) -> impl Iterator<Item = NodeRecord> + '_ {
436        self.swarm.peers().iter_peers()
437    }
438
439    /// Returns the number of peers in the peer set.
440    pub fn num_known_peers(&self) -> usize {
441        self.swarm.peers().num_known_peers()
442    }
443
444    /// Returns a new [`PeersHandle`] that can be cloned and shared.
445    ///
446    /// The [`PeersHandle`] can be used to interact with the network's peer set.
447    pub fn peers_handle(&self) -> PeersHandle {
448        self.swarm.peers().handle()
449    }
450
451    /// Collect the peers from the [`NetworkManager`] and write them to the given
452    /// `persistent_peers_file`.
453    ///
454    /// Only persists peers that are not currently backed off or banned. Includes metadata like
455    /// peer kind, fork ID, and reputation.
456    pub fn write_peers_to_file(&self, persistent_peers_file: &Path) -> Result<(), FsPathError> {
457        let peers = self.swarm.peers().persistable_peers().collect::<Vec<_>>();
458        persistent_peers_file.parent().map(fs::create_dir_all).transpose()?;
459        reth_fs_util::write_json_file(persistent_peers_file, &peers)?;
460        Ok(())
461    }
462
463    /// Returns a new [`FetchClient`] that can be cloned and shared.
464    ///
465    /// The [`FetchClient`] is the entrypoint for sending requests to the network, including
466    /// `snap/2` requests via its [`SnapClient`](reth_network_p2p::snap::client::SnapClient) impl.
467    pub fn fetch_client(&self) -> FetchClient<N> {
468        self.swarm.state().fetch_client()
469    }
470
471    /// Returns the current [`NetworkStatus`] for the local node.
472    pub fn status(&self) -> NetworkStatus {
473        let sessions = self.swarm.sessions();
474        let status = sessions.status();
475        let hello_message = sessions.hello_message();
476
477        #[expect(deprecated)]
478        NetworkStatus {
479            client_version: hello_message.client_version,
480            protocol_version: hello_message.protocol_version as u64,
481            eth_protocol_info: EthProtocolInfo {
482                difficulty: None,
483                head: status.blockhash,
484                network: status.chain.id(),
485                genesis: status.genesis,
486                config: Default::default(),
487            },
488            capabilities: hello_message
489                .protocols
490                .into_iter()
491                .map(|protocol| protocol.cap)
492                .collect(),
493        }
494    }
495
496    /// Sends an event to the [`TransactionsManager`](crate::transactions::TransactionsManager) if
497    /// configured.
498    fn notify_tx_manager(&self, event: NetworkTransactionEvent<N>) {
499        if let Some(ref tx) = self.to_transactions_manager &&
500            let Err(e) = tx.try_send(event)
501        {
502            match e {
503                TrySendError::Full(_) => {
504                    trace!(target: "net", "Transaction events channel at capacity, dropping event");
505                    self.metrics.total_dropped_tx_events_at_full_capacity.increment(1);
506                }
507                TrySendError::Closed(_) => {}
508            }
509        }
510    }
511
512    /// Sends an event to the [`EthRequestManager`](crate::eth_requests::EthRequestHandler) if
513    /// configured.
514    fn delegate_eth_request(&self, event: IncomingEthRequest<N>) {
515        if let Some(ref reqs) = self.to_eth_request_handler {
516            let _ = reqs.try_send(event).map_err(|e| {
517                if let TrySendError::Full(_) = e {
518                    debug!(target:"net", "EthRequestHandler channel is full!");
519                    self.metrics.total_dropped_eth_requests_at_full_capacity.increment(1);
520                }
521            });
522        }
523    }
524
525    /// Handle an incoming request from the peer
526    fn on_eth_request(&self, peer_id: PeerId, req: PeerRequest<N>) {
527        match req {
528            PeerRequest::GetBlockHeaders { request, response } => {
529                self.delegate_eth_request(IncomingEthRequest::GetBlockHeaders {
530                    peer_id,
531                    request,
532                    response,
533                })
534            }
535            PeerRequest::GetBlockBodies { request, response } => {
536                self.delegate_eth_request(IncomingEthRequest::GetBlockBodies {
537                    peer_id,
538                    request,
539                    response,
540                })
541            }
542            PeerRequest::GetNodeData { request, response } => {
543                self.delegate_eth_request(IncomingEthRequest::GetNodeData {
544                    peer_id,
545                    request,
546                    response,
547                })
548            }
549            PeerRequest::GetReceipts { request, response } => {
550                self.delegate_eth_request(IncomingEthRequest::GetReceipts {
551                    peer_id,
552                    request,
553                    response,
554                })
555            }
556            PeerRequest::GetReceipts69 { request, response } => {
557                self.delegate_eth_request(IncomingEthRequest::GetReceipts69 {
558                    peer_id,
559                    request,
560                    response,
561                })
562            }
563            PeerRequest::GetReceipts70 { request, response } => {
564                self.delegate_eth_request(IncomingEthRequest::GetReceipts70 {
565                    peer_id,
566                    request,
567                    response,
568                })
569            }
570            PeerRequest::GetBlockAccessLists { request, response } => {
571                self.delegate_eth_request(IncomingEthRequest::GetBlockAccessLists {
572                    peer_id,
573                    request,
574                    response,
575                })
576            }
577            PeerRequest::GetCells { request, response } => self
578                .delegate_eth_request(IncomingEthRequest::GetCells { peer_id, request, response }),
579            PeerRequest::GetPooledTransactions { request, response } => {
580                self.notify_tx_manager(NetworkTransactionEvent::GetPooledTransactions {
581                    peer_id,
582                    request,
583                    response,
584                });
585            }
586            PeerRequest::GetSnap { request, response } => self
587                .delegate_eth_request(IncomingEthRequest::GetSnap { peer_id, request, response }),
588        }
589    }
590
591    /// Invoked after a `NewBlock` message from the peer was validated
592    fn on_block_import_result(&mut self, event: BlockImportEvent<N::NewBlockPayload>) {
593        match event {
594            BlockImportEvent::Announcement(validation) => match validation {
595                BlockValidation::ValidHeader { block } => {
596                    self.swarm.state_mut().announce_new_block(block);
597                }
598                BlockValidation::ValidBlock { block } => {
599                    self.swarm.state_mut().announce_new_block_hash(block);
600                }
601            },
602            BlockImportEvent::Outcome(outcome) => {
603                let BlockImportOutcome { peer, result } = outcome;
604                match result {
605                    Ok(validated_block) => match validated_block {
606                        BlockValidation::ValidHeader { block } => {
607                            self.swarm.state_mut().update_peer_block(
608                                &peer,
609                                block.hash,
610                                block.number(),
611                            );
612                            self.swarm.state_mut().announce_new_block(block);
613                        }
614                        BlockValidation::ValidBlock { block } => {
615                            self.swarm.state_mut().announce_new_block_hash(block);
616                        }
617                    },
618                    Err(_err) => {
619                        self.swarm
620                            .state_mut()
621                            .peers_mut()
622                            .apply_reputation_change(&peer, ReputationChangeKind::BadBlock);
623                    }
624                }
625            }
626        }
627    }
628
629    /// Enforces [EIP-3675](https://eips.ethereum.org/EIPS/eip-3675#devp2p) consensus rules for the network protocol
630    ///
631    /// Depending on the mode of the network:
632    ///    - disconnect peer if in POS
633    ///    - execute the closure if in POW
634    fn within_pow_or_disconnect<F>(&mut self, peer_id: PeerId, only_pow: F)
635    where
636        F: FnOnce(&mut Self),
637    {
638        // reject message in POS
639        if self.handle.mode().is_stake() {
640            // connections to peers which send invalid messages should be terminated
641            self.swarm
642                .sessions_mut()
643                .disconnect(peer_id, Some(DisconnectReason::SubprotocolSpecific));
644        } else {
645            only_pow(self);
646        }
647    }
648
649    /// Handles a received Message from the peer's session.
650    fn on_peer_message(&mut self, peer_id: PeerId, msg: PeerMessage<N>) {
651        match msg {
652            PeerMessage::NewBlockHashes(hashes) => {
653                self.within_pow_or_disconnect(peer_id, |this| {
654                    // update peer's state, to track what blocks this peer has seen
655                    this.swarm.state_mut().on_new_block_hashes(peer_id, hashes.to_vec());
656                    // start block import process for the hashes
657                    this.block_import.on_new_block(peer_id, NewBlockEvent::Hashes(hashes));
658                })
659            }
660            PeerMessage::NewBlock(block) => {
661                self.within_pow_or_disconnect(peer_id, move |this| {
662                    this.swarm.state_mut().on_new_block(peer_id, block.hash);
663                    // start block import process
664                    this.block_import.on_new_block(peer_id, NewBlockEvent::Block(block));
665                });
666            }
667            PeerMessage::PooledTransactions(msg) => {
668                self.notify_tx_manager(NetworkTransactionEvent::IncomingPooledTransactionHashes {
669                    peer_id,
670                    msg,
671                });
672            }
673            PeerMessage::EthRequest(req) => {
674                self.on_eth_request(peer_id, req);
675            }
676            PeerMessage::ReceivedTransaction(msg) => {
677                self.notify_tx_manager(NetworkTransactionEvent::IncomingTransactions {
678                    peer_id,
679                    msg,
680                });
681            }
682            PeerMessage::SendTransactions(_) | PeerMessage::SendBroadcastPoolTransactions(_) => {
683                unreachable!("Not emitted by session")
684            }
685            PeerMessage::BlockRangeUpdated(_) => {}
686            PeerMessage::Other(other) => {
687                debug!(target: "net", message_id=%other.id, "Ignoring unsupported message");
688            }
689        }
690    }
691
692    /// Handler for received messages from a handle
693    fn on_handle_message(&mut self, msg: NetworkHandleMessage<N>) {
694        match msg {
695            NetworkHandleMessage::DiscoveryListener(tx) => {
696                self.swarm.state_mut().discovery_mut().add_listener(tx);
697            }
698            NetworkHandleMessage::AnnounceBlock(block, hash) => {
699                if self.handle.mode().is_stake() {
700                    // See [EIP-3675](https://eips.ethereum.org/EIPS/eip-3675#devp2p)
701                    warn!(target: "net", "Peer performed block propagation, but it is not supported in proof of stake (EIP-3675)");
702                    return
703                }
704                let msg = NewBlockMessage { hash, block: Arc::new(block) };
705                self.swarm.state_mut().announce_new_block(msg);
706            }
707            NetworkHandleMessage::EthRequest { peer_id, request } => {
708                self.swarm.sessions_mut().send_message(&peer_id, PeerMessage::EthRequest(request))
709            }
710            NetworkHandleMessage::SendTransaction { peer_id, msg } => {
711                self.swarm.sessions_mut().send_message(&peer_id, PeerMessage::SendTransactions(msg))
712            }
713            NetworkHandleMessage::SendBroadcastPoolTransactions { peer_id, msg } => self
714                .swarm
715                .sessions_mut()
716                .send_message(&peer_id, PeerMessage::SendBroadcastPoolTransactions(msg)),
717            NetworkHandleMessage::SendPooledTransactionHashes { peer_id, msg } => self
718                .swarm
719                .sessions_mut()
720                .send_message(&peer_id, PeerMessage::PooledTransactions(msg)),
721            NetworkHandleMessage::AddTrustedPeerId(peer_id) => {
722                self.swarm.state_mut().add_trusted_peer_id(peer_id);
723            }
724            NetworkHandleMessage::AddTrustedPeerNode(trusted_peer) => {
725                if !self.swarm.is_shutting_down() {
726                    self.swarm.state_mut().add_trusted_peer_node(trusted_peer);
727                }
728            }
729            NetworkHandleMessage::AddPeerAddress(peer, kind, addr) => {
730                // only add peer if we are not shutting down
731                if !self.swarm.is_shutting_down() {
732                    self.swarm.state_mut().add_peer_kind(peer, kind, addr);
733                }
734            }
735            NetworkHandleMessage::RemovePeer(peer_id, kind) => {
736                self.swarm.state_mut().remove_peer_kind(peer_id, kind);
737            }
738            NetworkHandleMessage::DisconnectPeer(peer_id, reason) => {
739                self.swarm.sessions_mut().disconnect(peer_id, reason);
740            }
741            NetworkHandleMessage::BanPeer(peer_id) => {
742                self.swarm.peers_mut().ban_peer_by_admin(peer_id);
743            }
744            NetworkHandleMessage::UnbanPeer(peer_id) => {
745                self.swarm.peers_mut().unban_peer_by_admin(peer_id);
746            }
747            NetworkHandleMessage::ConnectPeer(peer_id, kind, addr) => {
748                self.swarm.state_mut().add_and_connect(peer_id, kind, addr);
749            }
750            NetworkHandleMessage::SetNetworkState(net_state) => {
751                // Sets network connection state between Active and Hibernate.
752                // If hibernate stops the node to fill new outbound
753                // connections, this is beneficial for sync stages that do not require a network
754                // connection.
755                self.swarm.on_network_state_change(net_state);
756            }
757
758            NetworkHandleMessage::Shutdown(tx) => {
759                self.perform_network_shutdown();
760                let _ = tx.send(());
761            }
762            NetworkHandleMessage::ReputationChange(peer_id, kind) => {
763                self.swarm.peers_mut().apply_reputation_change(&peer_id, kind);
764            }
765            NetworkHandleMessage::GetReputationById(peer_id, tx) => {
766                let _ = tx.send(self.swarm.peers().get_reputation(&peer_id));
767            }
768            NetworkHandleMessage::FetchClient(tx) => {
769                let _ = tx.send(self.fetch_client());
770            }
771            NetworkHandleMessage::GetStatus(tx) => {
772                let _ = tx.send(self.status());
773            }
774            NetworkHandleMessage::StatusUpdate { head } => {
775                if let Some(transition) = self.swarm.sessions_mut().on_status_update(head) {
776                    self.swarm.state_mut().update_fork_id(transition.current);
777                }
778            }
779            NetworkHandleMessage::SetForkFilter { fork_filter } => {
780                let fork_id = self.swarm.sessions_mut().set_fork_filter(fork_filter);
781                self.swarm.state_mut().update_fork_id(fork_id);
782            }
783            NetworkHandleMessage::GetPeerInfos(tx) => {
784                let _ = tx.send(self.get_peer_infos());
785            }
786            NetworkHandleMessage::GetPeerInfoById(peer_id, tx) => {
787                let _ = tx.send(self.get_peer_info_by_id(peer_id));
788            }
789            NetworkHandleMessage::GetPeerInfosByIds(peer_ids, tx) => {
790                let _ = tx.send(self.get_peer_infos_by_ids(peer_ids));
791            }
792            NetworkHandleMessage::GetPeerInfosByPeerKind(kind, tx) => {
793                let peer_ids = self.swarm.peers().peers_by_kind(kind);
794                let _ = tx.send(self.get_peer_infos_by_ids(peer_ids));
795            }
796            NetworkHandleMessage::AddRlpxSubProtocol(proto) => self.add_rlpx_sub_protocol(proto),
797            NetworkHandleMessage::GetTransactionsHandle(tx) => {
798                if let Some(ref tx_inner) = self.to_transactions_manager {
799                    let _ = tx_inner.try_send(NetworkTransactionEvent::GetTransactionsHandle(tx));
800                } else {
801                    let _ = tx.send(None);
802                }
803            }
804            NetworkHandleMessage::InternalBlockRangeUpdate(block_range_update) => {
805                self.swarm.sessions_mut().update_advertised_block_range(block_range_update);
806            }
807            NetworkHandleMessage::EthMessage { peer_id, message } => {
808                self.swarm.sessions_mut().send_message(&peer_id, message)
809            }
810        }
811    }
812
813    fn on_swarm_event(&mut self, event: SwarmEvent<N>) {
814        // handle event
815        match event {
816            SwarmEvent::ValidMessage { peer_id, message } => self.on_peer_message(peer_id, message),
817            SwarmEvent::TcpListenerClosed { remote_addr } => {
818                trace!(target: "net", ?remote_addr, "TCP listener closed.");
819            }
820            SwarmEvent::TcpListenerError(err) => {
821                trace!(target: "net", %err, "TCP connection error.");
822            }
823            SwarmEvent::IncomingTcpConnection { remote_addr, session_id } => {
824                trace!(target: "net", ?session_id, ?remote_addr, "Incoming connection");
825                self.metrics.total_incoming_connections.increment(1);
826                self.metrics
827                    .incoming_connections
828                    .set(self.swarm.peers().num_inbound_connections() as f64);
829            }
830            SwarmEvent::OutgoingTcpConnection { remote_addr, peer_id } => {
831                trace!(target: "net", ?remote_addr, ?peer_id, "Starting outbound connection.");
832                self.metrics.total_outgoing_connections.increment(1);
833                self.update_pending_connection_metrics()
834            }
835            SwarmEvent::SessionEstablished {
836                peer_id,
837                remote_addr,
838                client_version,
839                capabilities,
840                version,
841                messages,
842                status,
843                direction,
844            } => {
845                let total_active = self.num_active_peers.fetch_add(1, Ordering::Relaxed) + 1;
846                self.metrics.connected_peers.set(total_active as f64);
847                debug!(
848                    target: "net",
849                    ?remote_addr,
850                    %client_version,
851                    ?peer_id,
852                    ?total_active,
853                    kind=%direction,
854                    peer_enode=%NodeRecord::new(remote_addr, peer_id),
855                    "Session established"
856                );
857
858                if direction.is_incoming() {
859                    self.swarm
860                        .state_mut()
861                        .peers_mut()
862                        .on_incoming_session_established(peer_id, remote_addr);
863                }
864
865                if direction.is_outgoing() {
866                    self.swarm.peers_mut().on_active_outgoing_established(peer_id);
867                }
868
869                self.update_active_connection_metrics();
870
871                let peer_kind = self
872                    .swarm
873                    .state()
874                    .peers()
875                    .peer_by_id(peer_id)
876                    .map(|(_, kind)| kind)
877                    .unwrap_or_default();
878                let session_info = SessionInfo {
879                    peer_id,
880                    remote_addr,
881                    client_version,
882                    capabilities,
883                    status,
884                    version,
885                    peer_kind,
886                };
887
888                self.event_sender
889                    .notify(NetworkEvent::ActivePeerSession { info: session_info, messages });
890            }
891            SwarmEvent::PeerAdded(peer_id) => {
892                trace!(target: "net", ?peer_id, "Peer added");
893                self.event_sender.notify(NetworkEvent::Peer(PeerEvent::PeerAdded(peer_id)));
894                self.metrics.tracked_peers.set(self.swarm.peers().num_known_peers() as f64);
895            }
896            SwarmEvent::PeerRemoved(peer_id) => {
897                trace!(target: "net", ?peer_id, "Peer dropped");
898                self.event_sender.notify(NetworkEvent::Peer(PeerEvent::PeerRemoved(peer_id)));
899                self.metrics.tracked_peers.set(self.swarm.peers().num_known_peers() as f64);
900            }
901            SwarmEvent::SessionClosed { peer_id, remote_addr, error } => {
902                let total_active = self.num_active_peers.fetch_sub(1, Ordering::Relaxed) - 1;
903                self.metrics.connected_peers.set(total_active as f64);
904                trace!(
905                    target: "net",
906                    ?remote_addr,
907                    ?peer_id,
908                    ?total_active,
909                    ?error,
910                    "Session disconnected"
911                );
912
913                // Capture direction before state is reset to Idle
914                let is_inbound = self.swarm.peers().is_inbound_peer(&peer_id);
915
916                let reason = if let Some(ref err) = error {
917                    // If the connection was closed due to an error, we report
918                    // the peer
919                    self.swarm.peers_mut().on_active_session_dropped(&remote_addr, &peer_id, err);
920                    self.backed_off_peers_metrics.increment_for_reason(
921                        BackoffReason::from_disconnect(err.as_disconnected()),
922                    );
923                    err.as_disconnected()
924                } else {
925                    // Gracefully disconnected
926                    self.swarm.peers_mut().on_active_session_gracefully_closed(peer_id);
927                    self.backed_off_peers_metrics
928                        .increment_for_reason(BackoffReason::GracefulClose);
929                    None
930                };
931                self.closed_sessions_metrics.active.increment(1);
932                self.update_active_connection_metrics();
933
934                if let Some(reason) = reason {
935                    if is_inbound {
936                        self.disconnect_metrics.increment_inbound(reason);
937                    } else {
938                        self.disconnect_metrics.increment_outbound(reason);
939                    }
940                }
941                self.metrics.backed_off_peers.set(self.swarm.peers().num_backed_off_peers() as f64);
942                self.event_sender
943                    .notify(NetworkEvent::Peer(PeerEvent::SessionClosed { peer_id, reason }));
944            }
945            SwarmEvent::IncomingPendingSessionClosed { remote_addr, error } => {
946                trace!(
947                    target: "net",
948                    ?remote_addr,
949                    ?error,
950                    "Incoming pending session failed"
951                );
952
953                if let Some(ref err) = error {
954                    self.swarm
955                        .state_mut()
956                        .peers_mut()
957                        .on_incoming_pending_session_dropped(remote_addr, err);
958                    self.pending_session_failure_metrics.inbound.increment(1);
959                    if let Some(reason) = err.as_disconnected() {
960                        self.disconnect_metrics.increment_inbound(reason);
961                    }
962                } else {
963                    self.swarm
964                        .state_mut()
965                        .peers_mut()
966                        .on_incoming_pending_session_gracefully_closed();
967                }
968                self.closed_sessions_metrics.incoming_pending.increment(1);
969                self.metrics
970                    .incoming_connections
971                    .set(self.swarm.peers().num_inbound_connections() as f64);
972            }
973            SwarmEvent::OutgoingPendingSessionClosed { remote_addr, peer_id, error } => {
974                trace!(
975                    target: "net",
976                    ?remote_addr,
977                    ?peer_id,
978                    ?error,
979                    "Outgoing pending session failed"
980                );
981
982                if let Some(ref err) = error {
983                    self.swarm.peers_mut().on_outgoing_pending_session_dropped(
984                        &remote_addr,
985                        &peer_id,
986                        err,
987                    );
988                    self.pending_session_failure_metrics.outbound.increment(1);
989                    self.backed_off_peers_metrics.increment_for_reason(
990                        BackoffReason::from_disconnect(err.as_disconnected()),
991                    );
992                    if let Some(reason) = err.as_disconnected() {
993                        self.disconnect_metrics.increment_outbound(reason);
994                    }
995                } else {
996                    self.swarm
997                        .state_mut()
998                        .peers_mut()
999                        .on_outgoing_pending_session_gracefully_closed(&peer_id);
1000                }
1001                self.closed_sessions_metrics.outgoing_pending.increment(1);
1002                self.update_pending_connection_metrics();
1003                self.metrics.backed_off_peers.set(self.swarm.peers().num_backed_off_peers() as f64);
1004            }
1005            SwarmEvent::OutgoingConnectionError { remote_addr, peer_id, error } => {
1006                trace!(
1007                    target: "net",
1008                    ?remote_addr,
1009                    ?peer_id,
1010                    %error,
1011                    "Outgoing connection error"
1012                );
1013
1014                self.swarm.peers_mut().on_outgoing_connection_failure(
1015                    &remote_addr,
1016                    &peer_id,
1017                    &error,
1018                );
1019
1020                self.backed_off_peers_metrics.increment_for_reason(BackoffReason::ConnectionError);
1021                self.metrics.backed_off_peers.set(self.swarm.peers().num_backed_off_peers() as f64);
1022                self.update_pending_connection_metrics();
1023            }
1024            SwarmEvent::BadMessage { peer_id } => {
1025                self.swarm
1026                    .state_mut()
1027                    .peers_mut()
1028                    .apply_reputation_change(&peer_id, ReputationChangeKind::BadMessage);
1029                self.metrics.invalid_messages_received.increment(1);
1030            }
1031            SwarmEvent::ProtocolBreach { peer_id } => {
1032                self.swarm
1033                    .state_mut()
1034                    .peers_mut()
1035                    .apply_reputation_change(&peer_id, ReputationChangeKind::BadProtocol);
1036            }
1037        }
1038    }
1039
1040    /// Returns [`PeerInfo`] for all connected peers
1041    fn get_peer_infos(&self) -> Vec<PeerInfo> {
1042        self.swarm
1043            .sessions()
1044            .active_sessions()
1045            .iter()
1046            .filter_map(|(&peer_id, session)| {
1047                self.swarm
1048                    .state()
1049                    .peers()
1050                    .peer_by_id(peer_id)
1051                    .map(|(record, kind)| session.peer_info(&record, kind))
1052            })
1053            .collect()
1054    }
1055
1056    /// Returns [`PeerInfo`] for a given peer.
1057    ///
1058    /// Returns `None` if there's no active session to the peer.
1059    fn get_peer_info_by_id(&self, peer_id: PeerId) -> Option<PeerInfo> {
1060        self.swarm.sessions().active_sessions().get(&peer_id).and_then(|session| {
1061            self.swarm
1062                .state()
1063                .peers()
1064                .peer_by_id(peer_id)
1065                .map(|(record, kind)| session.peer_info(&record, kind))
1066        })
1067    }
1068
1069    /// Returns [`PeerInfo`] for a given peers.
1070    ///
1071    /// Ignore the non-active peer.
1072    fn get_peer_infos_by_ids(&self, peer_ids: impl IntoIterator<Item = PeerId>) -> Vec<PeerInfo> {
1073        peer_ids.into_iter().filter_map(|peer_id| self.get_peer_info_by_id(peer_id)).collect()
1074    }
1075
1076    /// Updates the metrics for active,established connections
1077    #[inline]
1078    fn update_active_connection_metrics(&self) {
1079        self.metrics.incoming_connections.set(self.swarm.peers().num_inbound_connections() as f64);
1080        self.metrics.outgoing_connections.set(self.swarm.peers().num_outbound_connections() as f64);
1081    }
1082
1083    /// Updates the metrics for pending connections
1084    #[inline]
1085    fn update_pending_connection_metrics(&self) {
1086        self.metrics
1087            .pending_outgoing_connections
1088            .set(self.swarm.peers().num_pending_outbound_connections() as f64);
1089        self.metrics
1090            .total_pending_connections
1091            .set(self.swarm.sessions().num_pending_connections() as f64);
1092    }
1093
1094    /// Drives the [`NetworkManager`] future until a [`GracefulShutdown`] signal is received.
1095    ///
1096    /// This invokes the given function `shutdown_hook` while holding the graceful shutdown guard.
1097    pub async fn run_until_graceful_shutdown<F, R>(
1098        mut self,
1099        shutdown: GracefulShutdown,
1100        shutdown_hook: F,
1101    ) -> R
1102    where
1103        F: FnOnce(Self) -> R,
1104    {
1105        let mut graceful_guard = None;
1106        tokio::select! {
1107            _ = &mut self => {},
1108            guard = shutdown => {
1109                graceful_guard = Some(guard);
1110            },
1111        }
1112
1113        self.perform_network_shutdown();
1114        let res = shutdown_hook(self);
1115        drop(graceful_guard);
1116        res
1117    }
1118
1119    /// Performs a graceful network shutdown by stopping new connections from being accepted while
1120    /// draining current and pending connections.
1121    fn perform_network_shutdown(&mut self) {
1122        // Set connection status to `Shutdown`. Stops node from accepting
1123        // new incoming connections as well as sending connection requests to newly
1124        // discovered nodes.
1125        self.swarm.on_shutdown_requested();
1126        // Disconnect all active connections
1127        self.swarm.sessions_mut().disconnect_all(Some(DisconnectReason::ClientQuitting));
1128        // drop pending connections
1129        self.swarm.sessions_mut().disconnect_all_pending();
1130    }
1131}
1132
1133impl<N: NetworkPrimitives> Future for NetworkManager<N> {
1134    type Output = ();
1135
1136    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1137        let start = Instant::now();
1138        let mut poll_durations = NetworkManagerPollDurations::default();
1139
1140        let this = self.get_mut();
1141
1142        // poll new block imports (expected to be a noop for POS)
1143        while let Poll::Ready(outcome) = this.block_import.poll(cx) {
1144            this.on_block_import_result(outcome);
1145        }
1146
1147        // These loops drive the entire state of network and does a lot of work. Under heavy load
1148        // (many messages/events), data may arrive faster than it can be processed (incoming
1149        // messages/requests -> events), and it is possible that more data has already arrived by
1150        // the time an internal event is processed. Which could turn this loop into a busy loop.
1151        // Without yielding back to the executor, it can starve other tasks waiting on that
1152        // executor to execute them, or drive underlying resources To prevent this, we
1153        // preemptively return control when the `budget` is exhausted. The value itself is chosen
1154        // somewhat arbitrarily, it is high enough so the swarm can make meaningful progress but
1155        // low enough that this loop does not starve other tasks for too long. If the budget is
1156        // exhausted we manually yield back control to the (coop) scheduler. This manual yield
1157        // point should prevent situations where polling appears to be frozen. See also
1158        // <https://tokio.rs/blog/2020-04-preemption> And tokio's docs on cooperative scheduling
1159        // <https://docs.rs/tokio/latest/tokio/task/#cooperative-scheduling>
1160        //
1161        // Testing has shown that this loop naturally reaches the pending state within 1-5
1162        // iterations in << 100µs in most cases. On average it requires ~50µs, which is inside the
1163        // range of what's recommended as rule of thumb.
1164        // <https://ryhl.io/blog/async-what-is-blocking/>
1165
1166        // process incoming messages from a handle (`TransactionsManager` has one)
1167        //
1168        // will only be closed if the channel was deliberately closed since we always have an
1169        // instance of `NetworkHandle`
1170        let start_network_handle = Instant::now();
1171        let maybe_more_handle_messages = poll_nested_stream_with_budget!(
1172            "net",
1173            "Network message channel",
1174            DEFAULT_BUDGET_TRY_DRAIN_NETWORK_HANDLE_CHANNEL,
1175            this.from_handle_rx.poll_next_unpin(cx),
1176            |msg| this.on_handle_message(msg),
1177            error!("Network channel closed");
1178        );
1179        poll_durations.acc_network_handle = start_network_handle.elapsed();
1180
1181        // process incoming messages from the network
1182        let maybe_more_swarm_events = poll_nested_stream_with_budget!(
1183            "net",
1184            "Swarm events stream",
1185            DEFAULT_BUDGET_TRY_DRAIN_SWARM,
1186            this.swarm.poll_next_unpin(cx),
1187            |event| this.on_swarm_event(event),
1188        );
1189        poll_durations.acc_swarm =
1190            start_network_handle.elapsed() - poll_durations.acc_network_handle;
1191
1192        // all streams are fully drained and import futures pending
1193        if maybe_more_handle_messages || maybe_more_swarm_events {
1194            // make sure we're woken up again
1195            cx.waker().wake_by_ref();
1196            return Poll::Pending
1197        }
1198
1199        this.update_poll_metrics(start, poll_durations);
1200
1201        Poll::Pending
1202    }
1203}
1204
1205#[derive(Debug, Default)]
1206struct NetworkManagerPollDurations {
1207    acc_network_handle: Duration,
1208    acc_swarm: Duration,
1209}
reth_network/manager.rs

reth_network/
manager.rs
