reth_snap_sync/account/
download.rs

1//! Downloads account ranges in key order and commits each under the write it was fetched with.
2
3use crate::{
4    common::DownloadContext, AccountCoverage, SnapAccountStore, SnapAttemptStore, SnapSyncError,
5    SnapWrite, MAX_HASH,
6};
7use alloy_primitives::{map::B256Map, B256, U256};
8use reth_db_api::transaction::DbTxMut;
9use reth_downloaders::snap::{AccountRangeDownloader, AccountRangeOutcome, VerifiedAccountRange};
10use reth_eth_wire_types::snap::GetAccountRangeMessage;
11use reth_network_p2p::snap::client::SnapClient;
12use reth_network_peers::PeerId;
13use reth_storage_api::{
14    BlockHashReader, DBProvider, DatabaseProviderFactory, MetadataProvider, MetadataWriter,
15    StateWriter,
16};
17use reth_tasks::Runtime;
18use reth_trie_common::HashedStorage;
19use revm::bytecode::Bytecode;
20use std::fmt;
21
22/// Downloads the account ranges an attempt still needs, one at a time in key order.
23///
24/// [`Self::next`] fetches the range at the coverage cursor the store records; the caller resolves
25/// its storage and code, then [`Self::commit`] persists everything and moves the cursor.
26pub struct AccountRangeDownload<C, F> {
27    context: DownloadContext<C, F>,
28    // Coverage as last read from the store; none before the first request.
29    coverage: Option<AccountCoverage>,
30}
31
32impl<C, F> AccountRangeDownload<C, F> {
33    /// Creates a download that continues from the coverage the store records.
34    pub const fn new(client: C, factory: F, runtime: Runtime) -> Self {
35        Self { context: DownloadContext::new(client, factory, runtime), coverage: None }
36    }
37
38    /// Returns this download asking peers for at most `response_bytes` per response.
39    pub const fn with_response_bytes(mut self, response_bytes: u64) -> Self {
40        self.context.set_response_bytes(response_bytes);
41        self
42    }
43
44    /// How far the download has got, as last read from the store.
45    pub const fn coverage(&self) -> Option<AccountCoverage> {
46        self.coverage
47    }
48}
49
50impl<C, F> AccountRangeDownload<C, F>
51where
52    C: SnapClient + Clone + Unpin,
53    F: DatabaseProviderFactory + Clone + 'static,
54    F::Provider: MetadataProvider,
55    F::ProviderRW:
56        BlockHashReader + MetadataProvider + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
57{
58    /// Fetches the range at the coverage cursor.
59    ///
60    /// `Ok(None)` once every account is downloaded. A failed request leaves the cursor where it
61    /// was, so the download can be resumed from the persisted coverage.
62    pub async fn next(&mut self) -> Result<Option<AccountRangeStep>, SnapSyncError> {
63        let (write, root_hash, coverage) = self.active_write()?;
64        self.coverage = Some(coverage);
65        let Some(origin) = coverage.next() else { return Ok(None) };
66        self.request(write, root_hash, origin, MAX_HASH).await.map(Some)
67    }
68
69    /// Commits `verified` with the storage and code resolved for it, moving the cursor past it.
70    ///
71    /// Refused, with the coverage unchanged, when the attempt no longer accepts the write the
72    /// range was fetched under or a dependency is missing.
73    pub async fn commit(
74        &mut self,
75        verified: VerifiedRange,
76        storages: B256Map<HashedStorage>,
77        bytecodes: Vec<(B256, Bytecode)>,
78    ) -> Result<AccountCoverage, SnapSyncError> {
79        let coverage = self
80            .context
81            .commit(move |provider| {
82                let VerifiedRange { write, range } = verified;
83                provider.commit_account_range(write, &range, storages, bytecodes)
84            })
85            .await?;
86        self.coverage = Some(coverage);
87        Ok(coverage)
88    }
89
90    /// Fetches the first account scheduled for repair on its own, proved at the pivot.
91    ///
92    /// `Ok(None)` once nothing is scheduled.
93    pub async fn next_repair(&mut self) -> Result<Option<AccountRangeStep>, SnapSyncError> {
94        let (write, root_hash, _) = self.active_write()?;
95        let repairs = self.context.factory().database_provider_ro()?.snap_repairs(write)?;
96        let Some(hashed_address) = repairs.first() else { return Ok(None) };
97        self.request(write, root_hash, hashed_address, hashed_address).await.map(Some)
98    }
99
100    /// Commits the account `verified` repairs with the pivot's values of `slots`, returning how
101    /// many accounts remain scheduled.
102    pub async fn commit_repair(
103        &mut self,
104        verified: VerifiedRange,
105        slots: Vec<(B256, U256)>,
106    ) -> Result<usize, SnapSyncError> {
107        self.context
108            .commit(move |provider| {
109                let VerifiedRange { write, range } = verified;
110                provider.commit_account_repair(write, &range, slots)
111            })
112            .await
113    }
114
115    // Requests the accounts from `origin` through `limit` under `root_hash`.
116    async fn request(
117        &mut self,
118        write: SnapWrite,
119        root_hash: B256,
120        origin: B256,
121        limit: B256,
122    ) -> Result<AccountRangeStep, SnapSyncError> {
123        let request = GetAccountRangeMessage {
124            request_id: self.context.next_request_id(),
125            root_hash,
126            starting_hash: origin,
127            limit_hash: limit,
128            response_bytes: self.context.response_bytes(),
129        };
130        let downloader = AccountRangeDownloader::new(
131            self.context.client().clone(),
132            request,
133            self.context.runtime().clone(),
134        )?;
135
136        Ok(match downloader.await? {
137            AccountRangeOutcome::Verified(range) => {
138                AccountRangeStep::Verified(VerifiedRange { write, range })
139            }
140            AccountRangeOutcome::Unavailable { peer_id } => {
141                AccountRangeStep::Unavailable { origin, peer_id }
142            }
143        })
144    }
145
146    // The write the attempt accepts right now, the root to request against, and the coverage the
147    // store records for it.
148    fn active_write(&self) -> Result<(SnapWrite, B256, AccountCoverage), SnapSyncError> {
149        let provider = self.context.factory().database_provider_ro()?;
150        let write = provider.active_snap_write()?.ok_or(SnapSyncError::NoAttempt)?;
151        let root = provider.authorize_snap_write(write)?.state_root();
152        let coverage = provider.account_coverage(write)?.ok_or(SnapSyncError::NoCoverage)?;
153        Ok((write, root, coverage))
154    }
155}
156
157impl<C, F> fmt::Debug for AccountRangeDownload<C, F> {
158    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
159        f.debug_struct("AccountRangeDownload")
160            .field("context", &self.context)
161            .field("coverage", &self.coverage)
162            .finish()
163    }
164}
165
166/// What one request of an [`AccountRangeDownload`] produced.
167#[derive(Debug)]
168pub enum AccountRangeStep {
169    /// A range authenticated against the pivot root, waiting for its dependencies.
170    Verified(VerifiedRange),
171    /// The peer did not serve the range, so the cursor stays at `origin`.
172    Unavailable {
173        /// Key the range was requested from.
174        origin: B256,
175        /// Peer that answered without the state, so the retry can go elsewhere.
176        peer_id: PeerId,
177    },
178}
179
180/// A verified range with the write it was fetched under.
181///
182/// Only [`AccountRangeDownload::commit`] consumes it, so the range is committed under the
183/// attempt that was active when it was requested, never a later one.
184#[derive(Debug)]
185pub struct VerifiedRange {
186    // Write the attempt accepted when the range was requested.
187    write: SnapWrite,
188    // Accounts authenticated against that attempt's root.
189    range: VerifiedAccountRange,
190}
191
192impl VerifiedRange {
193    #[cfg(test)]
194    pub(crate) const fn new(write: SnapWrite, range: VerifiedAccountRange) -> Self {
195        Self { write, range }
196    }
197
198    /// Write the attempt accepted when the range was requested.
199    pub(crate) const fn write(&self) -> SnapWrite {
200        self.write
201    }
202
203    /// Key the range was requested from.
204    pub const fn origin(&self) -> B256 {
205        self.range.origin()
206    }
207
208    /// The accounts, whose storage and code the commit needs.
209    pub const fn range(&self) -> &VerifiedAccountRange {
210        &self.range
211    }
212}
213
214#[cfg(test)]
215mod tests {
216    use super::*;
217    use crate::{
218        test_utils::{
219            account, account_range, generation, hashed_factory, hashed_factory_with_roots, key,
220            state_root, verified_range, ScriptedSnapClient,
221        },
222        StateRepairs,
223    };
224    use reth_db_api::{cursor::DbCursorRO, tables, transaction::DbTx};
225    use reth_eth_wire_types::snap::AccountRangeMessage;
226    use reth_network_p2p::{error::PeerRequestResult, snap::client::SnapResponse};
227    use reth_network_peers::WithPeerId;
228    use reth_provider::{test_utils::MockNodeTypesWithDB, ProviderFactory};
229    use reth_trie_common::TrieAccount;
230    use std::sync::Arc;
231
232    const FAR: B256 = B256::repeat_byte(0xaa);
233
234    fn accounts() -> Vec<(B256, TrieAccount)> {
235        vec![(key(1), account(1)), (key(2), account(2)), (FAR, account(3))]
236    }
237
238    fn started(accounts: &[(B256, TrieAccount)]) -> ProviderFactory<MockNodeTypesWithDB> {
239        let factory = hashed_factory_with_roots(&[state_root(accounts); 3]);
240        let provider = factory.database_provider_rw().unwrap();
241        let write = provider.start_snap_attempt(generation(1, state_root(accounts))).unwrap();
242        provider.start_account_coverage(write).unwrap();
243        provider.commit().unwrap();
244        factory
245    }
246
247    type Download =
248        AccountRangeDownload<Arc<ScriptedSnapClient>, ProviderFactory<MockNodeTypesWithDB>>;
249
250    fn download(
251        responses: impl IntoIterator<Item = PeerRequestResult<SnapResponse>>,
252        factory: ProviderFactory<MockNodeTypesWithDB>,
253    ) -> (Arc<ScriptedSnapClient>, Download) {
254        let client = Arc::new(ScriptedSnapClient::new(responses));
255        let download = AccountRangeDownload::new(Arc::clone(&client), factory, Runtime::test());
256        (client, download)
257    }
258
259    fn stored_accounts(factory: &ProviderFactory<MockNodeTypesWithDB>) -> Vec<B256> {
260        let provider = factory.database_provider_ro().unwrap();
261        let mut cursor = provider.tx_ref().cursor_read::<tables::HashedAccounts>().unwrap();
262        cursor.walk(None).unwrap().map(|entry| entry.unwrap().0).collect()
263    }
264
265    async fn verified(download: &mut Download) -> VerifiedRange {
266        match download.next().await.unwrap().unwrap() {
267            AccountRangeStep::Verified(verified) => verified,
268            AccountRangeStep::Unavailable { .. } => panic!("fixture serves the range"),
269        }
270    }
271
272    #[tokio::test]
273    async fn downloads_the_trie_in_key_order_and_stops() {
274        let accounts = accounts();
275        let factory = started(&accounts);
276        let responses = [
277            // The first account only; the proof shows key 2 follows.
278            account_range(1, &accounts, 0..1, &[key(1)]),
279            // Everything from key 2 on.
280            account_range(2, &accounts, 1..3, &[key(2), FAR]),
281        ];
282        let (client, mut download) = download(responses, factory.clone());
283
284        let first = verified(&mut download).await;
285        assert_eq!(first.origin(), B256::ZERO);
286        assert_eq!(first.range().accounts().len(), 1);
287        let coverage = download.commit(first, Default::default(), Vec::new()).await.unwrap();
288        assert_eq!(coverage.next(), Some(key(2)));
289
290        let second = verified(&mut download).await;
291        assert_eq!(second.origin(), key(2));
292        let coverage = download.commit(second, Default::default(), Vec::new()).await.unwrap();
293        assert!(coverage.is_complete());
294
295        assert!(download.next().await.unwrap().is_none());
296        assert_eq!(*client.origins(), [B256::ZERO, key(2)]);
297        assert_eq!(stored_accounts(&factory), [key(1), key(2), FAR]);
298    }
299
300    #[tokio::test]
301    async fn an_unavailable_response_names_the_peer_and_leaves_the_cursor_in_place() {
302        let accounts = accounts();
303        let factory = started(&accounts);
304        let peer = PeerId::random();
305        let empty = AccountRangeMessage { request_id: 1, accounts: Vec::new(), proof: Vec::new() };
306        let responses = [
307            Ok(WithPeerId::new(peer, SnapResponse::AccountRange(empty))),
308            account_range(2, &accounts, 0..3, &[]),
309        ];
310        let (client, mut download) = download(responses, factory.clone());
311
312        let step = download.next().await.unwrap().unwrap();
313
314        assert!(matches!(
315            step,
316            AccountRangeStep::Unavailable { origin, peer_id }
317                if origin.is_zero() && peer_id == peer
318        ));
319        assert_eq!(download.coverage(), Some(AccountCoverage::START));
320
321        let range = verified(&mut download).await;
322        download.commit(range, Default::default(), Vec::new()).await.unwrap();
323        assert!(download.coverage().unwrap().is_complete());
324        assert_eq!(*client.origins(), [B256::ZERO, B256::ZERO]);
325    }
326
327    #[tokio::test]
328    async fn the_download_continues_from_the_coverage_the_store_records() {
329        let accounts = accounts();
330        let factory = started(&accounts);
331        // A previous run committed the first account.
332        let provider = factory.database_provider_rw().unwrap();
333        let write = provider.active_snap_write().unwrap().unwrap();
334        let head = verified_range(&accounts, 0..1, B256::ZERO, &[key(1)]);
335        provider.commit_account_range(write, &head, Default::default(), Vec::new()).unwrap();
336        provider.commit().unwrap();
337        let (client, mut download) =
338            download([account_range(1, &accounts, 1..3, &[key(2), FAR])], factory.clone());
339
340        let rest = verified(&mut download).await;
341
342        assert_eq!(rest.origin(), key(2));
343        assert_eq!(*client.origins(), [key(2)]);
344        download.commit(rest, Default::default(), Vec::new()).await.unwrap();
345        assert!(download.coverage().unwrap().is_complete());
346        assert_eq!(stored_accounts(&factory), [key(1), key(2), FAR]);
347    }
348
349    #[tokio::test]
350    async fn a_range_fetched_under_a_replaced_attempt_is_refused_at_commit() {
351        let accounts = accounts();
352        let factory = started(&accounts);
353        let (_, mut download) = download([account_range(1, &accounts, 0..3, &[])], factory.clone());
354        let range = verified(&mut download).await;
355        // The attempt is replaced after the range was fetched.
356        let provider = factory.database_provider_rw().unwrap();
357        provider.start_snap_attempt(generation(1, state_root(&accounts))).unwrap();
358        provider.commit().unwrap();
359
360        let error = download.commit(range, Default::default(), Vec::new()).await.unwrap_err();
361
362        assert!(matches!(error, SnapSyncError::StaleWrite { .. }));
363        assert!(stored_accounts(&factory).is_empty());
364        assert_eq!(download.coverage(), Some(AccountCoverage::START));
365    }
366
367    #[tokio::test]
368    async fn nothing_is_requested_without_an_active_attempt() {
369        let (client, mut download) = download([], hashed_factory());
370
371        assert!(matches!(download.next().await, Err(SnapSyncError::NoAttempt)));
372        assert!(client.origins().is_empty());
373        assert_eq!(download.coverage(), None);
374    }
375
376    #[tokio::test]
377    async fn nothing_is_requested_without_recorded_coverage() {
378        let factory = hashed_factory_with_roots(&[state_root(&accounts())]);
379        let provider = factory.database_provider_rw().unwrap();
380        provider.start_snap_attempt(generation(1, state_root(&accounts()))).unwrap();
381        provider.commit().unwrap();
382        let (client, mut download) = download([], factory);
383
384        assert!(matches!(download.next().await, Err(SnapSyncError::NoCoverage)));
385        assert!(client.origins().is_empty());
386    }
387
388    #[tokio::test]
389    async fn repairs_are_fetched_one_account_at_a_time() {
390        let accounts = accounts();
391        let factory = started(&accounts);
392        let responses = [account_range(1, &accounts, 1..2, &[key(2)])];
393        let (client, mut download) = download(responses, factory.clone());
394        assert!(download.next_repair().await.unwrap().is_none());
395
396        let provider = factory.database_provider_rw().unwrap();
397        let write = provider.active_snap_write().unwrap().unwrap();
398        let mut repairs = StateRepairs::default();
399        repairs.insert_account(key(2));
400        provider.schedule_snap_repairs(write, repairs).unwrap();
401        provider.commit().unwrap();
402
403        let Some(AccountRangeStep::Verified(range)) = download.next_repair().await.unwrap() else {
404            panic!("fixture serves the account")
405        };
406        assert_eq!(*client.origins(), [key(2)]);
407        assert_eq!(range.range().accounts(), &accounts[1..2]);
408    }
409}
reth_snap_sync/account/download.rs

reth_snap_sync/account/
download.rs
