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, BodiesFut},
15    download::DownloadClient,
16    error::{PeerRequestResult, RequestError},
17    headers::client::{HeadersClient, HeadersRequest},
18    priority::Priority,
19    receipts::client::{ReceiptsClient, ReceiptsFut},
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    ) -> std::pin::Pin<Box<dyn Future<Output = PeerRequestResult<SnapResponse>> + Send + Sync>>
68    {
69        let (response, rx) = oneshot::channel();
70        if self.request_tx.send(DownloadRequest::GetSnap { request, response, priority }).is_ok() {
71            Box::pin(FlattenedResponse::from(rx))
72        } else {
73            Box::pin(future::err(RequestError::ChannelClosed))
74        }
75    }
76}
77
78// The `Output` future of the [HeadersClient] impl of [FetchClient] that either returns a response
79// or an error.
80type HeadersClientFuture<T> = Either<FlattenedResponse<T>, future::Ready<T>>;
81
82impl<N: NetworkPrimitives> HeadersClient for FetchClient<N> {
83    type Header = N::BlockHeader;
84    type Output = HeadersClientFuture<PeerRequestResult<Vec<N::BlockHeader>>>;
85
86    /// Sends a `GetBlockHeaders` request to an available peer.
87    fn get_headers_with_priority(
88        &self,
89        request: HeadersRequest,
90        priority: Priority,
91    ) -> Self::Output {
92        let (response, rx) = oneshot::channel();
93        if self
94            .request_tx
95            .send(DownloadRequest::GetBlockHeaders { request, response, priority })
96            .is_ok()
97        {
98            Either::Left(FlattenedResponse::from(rx))
99        } else {
100            Either::Right(future::err(RequestError::ChannelClosed))
101        }
102    }
103}
104
105impl<N: NetworkPrimitives> BodiesClient for FetchClient<N> {
106    type Body = N::BlockBody;
107    type Output = BodiesFut<N::BlockBody>;
108
109    /// Sends a `GetBlockBodies` request to an available peer.
110    fn get_block_bodies_with_priority_and_range_hint(
111        &self,
112        request: Vec<B256>,
113        priority: Priority,
114        range_hint: Option<RangeInclusive<u64>>,
115    ) -> Self::Output {
116        let (response, rx) = oneshot::channel();
117        if self
118            .request_tx
119            .send(DownloadRequest::GetBlockBodies { request, response, priority, range_hint })
120            .is_ok()
121        {
122            Box::pin(FlattenedResponse::from(rx))
123        } else {
124            Box::pin(future::err(RequestError::ChannelClosed))
125        }
126    }
127}
128
129impl<N: NetworkPrimitives> ReceiptsClient for FetchClient<N> {
130    type Receipt = N::Receipt;
131    type Output = ReceiptsFut<N::Receipt>;
132
133    fn get_receipts_with_priority(&self, request: Vec<B256>, priority: Priority) -> Self::Output {
134        let (response, rx) = oneshot::channel();
135        if self
136            .request_tx
137            .send(DownloadRequest::GetReceipts { request, response, priority })
138            .is_ok()
139        {
140            Box::pin(FlattenedResponse::from(rx))
141        } else {
142            Box::pin(future::err(RequestError::ChannelClosed))
143        }
144    }
145}
146
147impl<N: NetworkPrimitives> BlockClient for FetchClient<N> {
148    type Block = N::Block;
149}
150
151impl<N: NetworkPrimitives> BlockAccessListsClient for FetchClient<N> {
152    type Output =
153        std::pin::Pin<Box<dyn Future<Output = PeerRequestResult<BlockAccessLists>> + Send + Sync>>;
154
155    fn get_block_access_lists_with_priority_and_requirement(
156        &self,
157        hashes: Vec<B256>,
158        priority: Priority,
159        requirement: BalRequirement,
160    ) -> Self::Output {
161        let (response, rx) = oneshot::channel();
162        if self
163            .request_tx
164            .send(DownloadRequest::GetBlockAccessLists {
165                request: hashes,
166                response,
167                priority,
168                requirement,
169            })
170            .is_ok()
171        {
172            Box::pin(FlattenedResponse::from(rx))
173        } else {
174            Box::pin(future::err(RequestError::ChannelClosed))
175        }
176    }
177}
178
179impl<N: NetworkPrimitives> SnapClient for FetchClient<N> {
180    type Output =
181        std::pin::Pin<Box<dyn Future<Output = PeerRequestResult<SnapResponse>> + Send + Sync>>;
182
183    /// Sends a `GetAccountRange` (`snap/2`) request to an available peer.
184    fn get_account_range_with_priority(
185        &self,
186        request: GetAccountRangeMessage,
187        priority: Priority,
188    ) -> Self::Output {
189        self.send_snap_request(SnapProtocolMessage::GetAccountRange(request), priority)
190    }
191
192    /// Sends a `GetStorageRanges` (`snap/2`) request to an available peer.
193    fn get_storage_ranges(&self, request: GetStorageRangesMessage) -> Self::Output {
194        self.get_storage_ranges_with_priority(request, Priority::Normal)
195    }
196
197    /// Sends a `GetStorageRanges` (`snap/2`) request to an available peer.
198    fn get_storage_ranges_with_priority(
199        &self,
200        request: GetStorageRangesMessage,
201        priority: Priority,
202    ) -> Self::Output {
203        self.send_snap_request(SnapProtocolMessage::GetStorageRanges(request), priority)
204    }
205
206    /// Sends a `GetByteCodes` (`snap/2`) request to an available peer.
207    fn get_byte_codes(&self, request: GetByteCodesMessage) -> Self::Output {
208        self.get_byte_codes_with_priority(request, Priority::Normal)
209    }
210
211    /// Sends a `GetByteCodes` (`snap/2`) request to an available peer.
212    fn get_byte_codes_with_priority(
213        &self,
214        request: GetByteCodesMessage,
215        priority: Priority,
216    ) -> Self::Output {
217        self.send_snap_request(SnapProtocolMessage::GetByteCodes(request), priority)
218    }
219
220    /// Sends a `GetBlockAccessLists` (`snap/2`) request to an available peer.
221    fn get_block_access_lists_with_priority(
222        &self,
223        request: GetBlockAccessListsMessage,
224        priority: Priority,
225    ) -> Self::Output {
226        self.send_snap_request(SnapProtocolMessage::GetBlockAccessLists(request), priority)
227    }
228}
reth_network/fetch/client.rs

reth_network/fetch/
client.rs
