reth_network/fetch/
client.rs

1//! A client implementation that can interact with the network and download data.
2
3use crate::{fetch::DownloadRequest, flattened_response::FlattenedResponse};
4use alloy_primitives::B256;
5use futures::{future, future::Either};
6use reth_eth_wire::{BlockAccessLists, EthNetworkPrimitives, NetworkPrimitives};
7use reth_eth_wire_types::snap::{
8    GetAccountRangeMessage, GetBlockAccessListsMessage, GetByteCodesMessage,
9    GetStorageRangesMessage, SnapProtocolMessage,
10};
11use reth_network_api::test_utils::PeersHandle;
12use reth_network_p2p::{
13    block_access_lists::client::{BalRequirement, BlockAccessListsClient},
14    bodies::client::BodiesClient,
15    download::DownloadClient,
16    error::{PeerRequestResult, RequestError},
17    headers::client::{HeadersClient, HeadersRequest},
18    priority::Priority,
19    receipts::client::{ReceiptsClient, ReceiptsResponse},
20    snap::client::{SnapClient, SnapResponse},
21    BlockClient,
22};
23use reth_network_peers::PeerId;
24use reth_network_types::ReputationChangeKind;
25use std::{
26    ops::RangeInclusive,
27    sync::{
28        atomic::{AtomicUsize, Ordering},
29        Arc,
30    },
31};
32use tokio::sync::{mpsc::UnboundedSender, oneshot};
33
34#[cfg_attr(doc, aquamarine::aquamarine)]
35/// Front-end API for fetching data from the network.
36///
37/// Following diagram illustrates how a request, See [`HeadersClient::get_headers`] and
38/// [`BodiesClient::get_block_bodies`] is handled internally.
39///
40/// include_mmd!("docs/mermaid/fetch-client.mmd")
41#[derive(Debug, Clone)]
42pub struct FetchClient<N: NetworkPrimitives = EthNetworkPrimitives> {
43    /// Sender half of the request channel.
44    pub(crate) request_tx: UnboundedSender<DownloadRequest<N>>,
45    /// The handle to the peers
46    pub(crate) peers_handle: PeersHandle,
47    /// Number of active peer sessions the node's currently handling.
48    pub(crate) num_active_peers: Arc<AtomicUsize>,
49}
50
51impl<N: NetworkPrimitives> DownloadClient for FetchClient<N> {
52    fn report_bad_message(&self, peer_id: PeerId) {
53        self.peers_handle.reputation_change(peer_id, ReputationChangeKind::BadMessage);
54    }
55
56    fn num_connected_peers(&self) -> usize {
57        self.num_active_peers.load(Ordering::Relaxed)
58    }
59}
60
61impl<N: NetworkPrimitives> FetchClient<N> {
62    /// Sends a `snap/2` request to an available peer.
63    fn send_snap_request(
64        &self,
65        request: SnapProtocolMessage,
66        priority: Priority,
67    ) -> FetchClientFuture<PeerRequestResult<SnapResponse>> {
68        let (response, rx) = oneshot::channel();
69        if self.request_tx.send(DownloadRequest::GetSnap { request, response, priority }).is_ok() {
70            Either::Left(FlattenedResponse::from(rx))
71        } else {
72            Either::Right(future::err(RequestError::ChannelClosed))
73        }
74    }
75}
76
77/// A request future that either waits for a network response or reports a closed request channel.
78type FetchClientFuture<T> = Either<FlattenedResponse<T>, future::Ready<T>>;
79
80impl<N: NetworkPrimitives> HeadersClient for FetchClient<N> {
81    type Header = N::BlockHeader;
82    type Output = FetchClientFuture<PeerRequestResult<Vec<N::BlockHeader>>>;
83
84    /// Sends a `GetBlockHeaders` request to an available peer.
85    fn get_headers_with_priority(
86        &self,
87        request: HeadersRequest,
88        priority: Priority,
89    ) -> Self::Output {
90        let (response, rx) = oneshot::channel();
91        if self
92            .request_tx
93            .send(DownloadRequest::GetBlockHeaders { request, response, priority })
94            .is_ok()
95        {
96            Either::Left(FlattenedResponse::from(rx))
97        } else {
98            Either::Right(future::err(RequestError::ChannelClosed))
99        }
100    }
101}
102
103impl<N: NetworkPrimitives> BodiesClient for FetchClient<N> {
104    type Body = N::BlockBody;
105    type Output = FetchClientFuture<PeerRequestResult<Vec<N::BlockBody>>>;
106
107    /// Sends a `GetBlockBodies` request to an available peer.
108    fn get_block_bodies_with_priority_and_range_hint(
109        &self,
110        request: Vec<B256>,
111        priority: Priority,
112        range_hint: Option<RangeInclusive<u64>>,
113    ) -> Self::Output {
114        let (response, rx) = oneshot::channel();
115        if self
116            .request_tx
117            .send(DownloadRequest::GetBlockBodies { request, response, priority, range_hint })
118            .is_ok()
119        {
120            Either::Left(FlattenedResponse::from(rx))
121        } else {
122            Either::Right(future::err(RequestError::ChannelClosed))
123        }
124    }
125}
126
127impl<N: NetworkPrimitives> ReceiptsClient for FetchClient<N> {
128    type Receipt = N::Receipt;
129    type Output = FetchClientFuture<PeerRequestResult<ReceiptsResponse<N::Receipt>>>;
130
131    fn get_receipts_with_priority(&self, request: Vec<B256>, priority: Priority) -> Self::Output {
132        let (response, rx) = oneshot::channel();
133        if self
134            .request_tx
135            .send(DownloadRequest::GetReceipts { request, response, priority })
136            .is_ok()
137        {
138            Either::Left(FlattenedResponse::from(rx))
139        } else {
140            Either::Right(future::err(RequestError::ChannelClosed))
141        }
142    }
143}
144
145impl<N: NetworkPrimitives> BlockClient for FetchClient<N> {
146    type Block = N::Block;
147}
148
149impl<N: NetworkPrimitives> BlockAccessListsClient for FetchClient<N> {
150    type Output = FetchClientFuture<PeerRequestResult<BlockAccessLists>>;
151
152    fn get_block_access_lists_with_priority_and_requirement(
153        &self,
154        hashes: Vec<B256>,
155        priority: Priority,
156        requirement: BalRequirement,
157    ) -> Self::Output {
158        let (response, rx) = oneshot::channel();
159        if self
160            .request_tx
161            .send(DownloadRequest::GetBlockAccessLists {
162                request: hashes,
163                response,
164                priority,
165                requirement,
166            })
167            .is_ok()
168        {
169            Either::Left(FlattenedResponse::from(rx))
170        } else {
171            Either::Right(future::err(RequestError::ChannelClosed))
172        }
173    }
174}
175
176impl<N: NetworkPrimitives> SnapClient for FetchClient<N> {
177    type Output = FetchClientFuture<PeerRequestResult<SnapResponse>>;
178
179    /// Sends a `GetAccountRange` (`snap/2`) request to an available peer.
180    fn get_account_range_with_priority(
181        &self,
182        request: GetAccountRangeMessage,
183        priority: Priority,
184    ) -> Self::Output {
185        self.send_snap_request(SnapProtocolMessage::GetAccountRange(request), priority)
186    }
187
188    /// Sends a `GetStorageRanges` (`snap/2`) request to an available peer.
189    fn get_storage_ranges(&self, request: GetStorageRangesMessage) -> Self::Output {
190        self.get_storage_ranges_with_priority(request, Priority::Normal)
191    }
192
193    /// Sends a `GetStorageRanges` (`snap/2`) request to an available peer.
194    fn get_storage_ranges_with_priority(
195        &self,
196        request: GetStorageRangesMessage,
197        priority: Priority,
198    ) -> Self::Output {
199        self.send_snap_request(SnapProtocolMessage::GetStorageRanges(request), priority)
200    }
201
202    /// Sends a `GetByteCodes` (`snap/2`) request to an available peer.
203    fn get_byte_codes(&self, request: GetByteCodesMessage) -> Self::Output {
204        self.get_byte_codes_with_priority(request, Priority::Normal)
205    }
206
207    /// Sends a `GetByteCodes` (`snap/2`) request to an available peer.
208    fn get_byte_codes_with_priority(
209        &self,
210        request: GetByteCodesMessage,
211        priority: Priority,
212    ) -> Self::Output {
213        self.send_snap_request(SnapProtocolMessage::GetByteCodes(request), priority)
214    }
215
216    /// Sends a `GetBlockAccessLists` (`snap/2`) request to an available peer.
217    fn get_block_access_lists_with_priority(
218        &self,
219        request: GetBlockAccessListsMessage,
220        priority: Priority,
221    ) -> Self::Output {
222        self.send_snap_request(SnapProtocolMessage::GetBlockAccessLists(request), priority)
223    }
224}
reth_network/fetch/client.rs

reth_network/fetch/
client.rs
