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

reth_snap_sync/common/
download.rs
