reth_snap_sync/common/
download.rs

1//! Client, database and request settings shared by the account, storage and bytecode downloads.
2//!
3//! Each download holds a [`DownloadContext`] to number its requests and commit verified responses
4//! in one transaction on the blocking pool.
5
6use crate::SnapSyncError;
7use alloy_primitives::B256;
8use reth_storage_api::{DBProvider, DatabaseProviderFactory};
9use reth_storage_errors::provider::ProviderError;
10use reth_tasks::Runtime;
11use std::fmt;
12
13/// Default soft response limit for snap requests, matching common peer limits.
14pub const DEFAULT_RESPONSE_BYTES: u64 = 512 * 1024;
15
16/// Inclusive upper bound covering the full trie keyspace.
17pub const MAX_HASH: B256 = B256::new([0xff; B256::len_bytes()]);
18
19/// Client, database and request settings a domain download sends and commits through.
20pub(crate) struct DownloadContext<C, F> {
21    client: C,
22    factory: F,
23    // Proof verification and commits run on the blocking pool.
24    runtime: Runtime,
25    response_bytes: u64,
26    // Distinguishes responses to reissued requests.
27    request_id: u64,
28}
29
30impl<C, F> DownloadContext<C, F> {
31    pub(crate) const fn new(client: C, factory: F, runtime: Runtime) -> Self {
32        Self { client, factory, runtime, response_bytes: DEFAULT_RESPONSE_BYTES, request_id: 0 }
33    }
34
35    pub(crate) const fn client(&self) -> &C {
36        &self.client
37    }
38
39    pub(crate) const fn factory(&self) -> &F {
40        &self.factory
41    }
42
43    pub(crate) const fn runtime(&self) -> &Runtime {
44        &self.runtime
45    }
46
47    pub(crate) const fn response_bytes(&self) -> u64 {
48        self.response_bytes
49    }
50
51    pub(crate) const fn set_response_bytes(&mut self, response_bytes: u64) {
52        self.response_bytes = response_bytes;
53    }
54
55    /// Returns the id for the next request.
56    pub(crate) const fn next_request_id(&mut self) -> u64 {
57        self.request_id = self.request_id.wrapping_add(1);
58        self.request_id
59    }
60}
61
62impl<C, F> DownloadContext<C, F>
63where
64    F: DatabaseProviderFactory + Clone + 'static,
65    F::ProviderRW: DBProvider,
66{
67    /// Runs `read` on the blocking pool, where scans touching many keys belong.
68    pub(crate) async fn read<T: Send + 'static>(
69        &self,
70        read: impl FnOnce(&F::Provider) -> Result<T, SnapSyncError> + Send + 'static,
71    ) -> Result<T, SnapSyncError> {
72        let factory = self.factory.clone();
73        self.runtime
74            .spawn_blocking(move || read(&factory.database_provider_ro()?))
75            .await
76            .map_err(|error| SnapSyncError::Provider(ProviderError::other(error)))?
77    }
78
79    /// Runs `write` in one transaction on the blocking pool, committing only if it succeeds.
80    pub(crate) async fn commit<T: Send + 'static>(
81        &self,
82        write: impl FnOnce(&F::ProviderRW) -> Result<T, SnapSyncError> + Send + 'static,
83    ) -> Result<T, SnapSyncError> {
84        let factory = self.factory.clone();
85        self.runtime
86            .spawn_blocking(move || -> Result<T, SnapSyncError> {
87                let provider = factory.database_provider_rw()?;
88                let output = write(&provider)?;
89                provider.commit()?;
90                Ok(output)
91            })
92            .await
93            .map_err(|error| SnapSyncError::Provider(ProviderError::other(error)))?
94    }
95}
96
97impl<C, F> fmt::Debug for DownloadContext<C, F> {
98    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
99        f.debug_struct("DownloadContext")
100            .field("response_bytes", &self.response_bytes)
101            .field("request_id", &self.request_id)
102            .finish_non_exhaustive()
103    }
104}
reth_snap_sync/common/download.rs

reth_snap_sync/common/
download.rs
