reth_snap_sync/storage/
download.rs

1//! Downloads the storage of an account range's contracts, committing every verified response.
2
3use crate::{
4    common::DownloadContext, SnapAccountStore, SnapStorageStore, SnapSyncError, StorageChunk,
5    StorageProgress, VerifiedRange, MAX_HASH,
6};
7use alloy_primitives::{B256, U256};
8use futures::future::join_all;
9use reth_db_api::transaction::DbTxMut;
10use reth_downloaders::snap::{
11    StorageRangeDownloader, StorageRangeOutcome, VerifiedAccountBatch, VerifiedStorageRanges,
12};
13use reth_eth_wire_types::snap::GetStorageRangesMessage;
14use reth_network_p2p::snap::client::SnapClient;
15use reth_network_peers::PeerId;
16use reth_storage_api::{
17    DBProvider, DatabaseProviderFactory, MetadataProvider, MetadataWriter, StateWriter,
18};
19use reth_tasks::Runtime;
20use std::fmt;
21
22/// Default number of contracts asked for per storage request.
23pub const DEFAULT_STORAGE_ACCOUNTS: usize = 128;
24
25/// Default number of scheduled slots fetched again per repair batch.
26pub const DEFAULT_REPAIR_SLOTS: usize = 128;
27
28/// Downloads the storage an account range's contracts still need, and repaired slots again.
29///
30/// Each range response is committed before the next request, so at most one is held however large
31/// a contract is, and a download resumes from the persisted progress.
32pub struct StorageRangeDownload<C, F> {
33    context: DownloadContext<C, F>,
34    // Contracts asked for per request.
35    max_accounts: usize,
36    // Scheduled slots fetched again per repair batch.
37    max_repair_slots: usize,
38}
39
40impl<C, F> StorageRangeDownload<C, F> {
41    /// Creates a download that continues from the progress the store records.
42    pub const fn new(client: C, factory: F, runtime: Runtime) -> Self {
43        Self {
44            context: DownloadContext::new(client, factory, runtime),
45            max_accounts: DEFAULT_STORAGE_ACCOUNTS,
46            max_repair_slots: DEFAULT_REPAIR_SLOTS,
47        }
48    }
49
50    /// Returns this download asking peers for at most `response_bytes` per response.
51    pub const fn with_response_bytes(mut self, response_bytes: u64) -> Self {
52        self.context.set_response_bytes(response_bytes);
53        self
54    }
55
56    /// Returns this download asking for at most `max_accounts` contracts per request, at least one.
57    pub const fn with_max_accounts(mut self, max_accounts: usize) -> Self {
58        self.max_accounts = if max_accounts == 0 { 1 } else { max_accounts };
59        self
60    }
61
62    /// Returns this download fetching at most `max_repair_slots` scheduled slots again per repair
63    /// batch, at least one.
64    pub const fn with_max_repair_slots(mut self, max_repair_slots: usize) -> Self {
65        self.max_repair_slots = if max_repair_slots == 0 { 1 } else { max_repair_slots };
66        self
67    }
68}
69
70impl<C, F> StorageRangeDownload<C, F>
71where
72    C: SnapClient + Clone + Unpin,
73    F: DatabaseProviderFactory + Clone + 'static,
74    F::Provider: MetadataProvider,
75    F::ProviderRW: MetadataProvider + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
76{
77    /// Requests storage `range` still needs and commits the verified response.
78    ///
79    /// [`StorageRangeStep::Complete`] once every contract in the range is persisted, so the range
80    /// can commit without supplying their storage. A failed request leaves the progress in place.
81    pub async fn next(&mut self, range: &VerifiedRange) -> Result<StorageRangeStep, SnapSyncError> {
82        let (write, origin) = (range.write(), range.origin());
83        let progress =
84            self.context.factory().database_provider_ro()?.storage_progress(write, origin)?;
85        let contracts = range.range().storage_batch();
86        let Some(first) =
87            contracts.accounts().iter().position(|(account, _)| !progress.is_complete(*account))
88        else {
89            return Ok(StorageRangeStep::Complete)
90        };
91        let end = progress.request_end(contracts.accounts(), first, self.max_accounts);
92        let batch = contracts.range(first..end).expect("positions are inside the batch");
93        let from = progress.resume_at(batch.accounts()[0].0).expect("first contract is incomplete");
94
95        let request = GetStorageRangesMessage {
96            request_id: self.context.next_request_id(),
97            root_hash: batch.state_root(),
98            account_hashes: batch.accounts().iter().map(|(account, _)| *account).collect(),
99            starting_hash: from.into(),
100            limit_hash: MAX_HASH.into(),
101            response_bytes: self.context.response_bytes(),
102        };
103        let downloader = StorageRangeDownloader::new(
104            self.context.client().clone(),
105            request,
106            &batch,
107            self.context.runtime().clone(),
108        )?;
109        let ranges = match downloader.await? {
110            StorageRangeOutcome::Verified(ranges) => ranges,
111            StorageRangeOutcome::Unavailable { peer_id } => {
112                return Ok(StorageRangeStep::Unavailable { peer_id })
113            }
114        };
115        let chunks = chunks(ranges, batch, from)?;
116        // One response's chunks commit together.
117        let committed = self
118            .context
119            .commit(move |provider| {
120                let mut progress = StorageProgress::START;
121                for chunk in chunks {
122                    progress = provider.commit_storage_chunk(write, origin, chunk)?;
123                }
124                Ok(progress)
125            })
126            .await?;
127        Ok(StorageRangeStep::Committed(committed))
128    }
129
130    /// Fetches the pivot's values of the next batch of slots scheduled for repair at `range`'s
131    /// origin concurrently, zero where its storage holds none. Unserved slots stay scheduled, and
132    /// `Ok(None)` when none was served.
133    pub async fn repair_slots(
134        &mut self,
135        range: &VerifiedRange,
136    ) -> Result<Option<Vec<(B256, U256)>>, SnapSyncError> {
137        let (write, account) = (range.write(), range.origin());
138        let repairs = self.context.factory().database_provider_ro()?.snap_repairs(write)?;
139        // Only a contract the pivot holds has slots to fetch; for any other account they are zero.
140        let contracts = range.range().storage_batch();
141        let Some(batch) = contracts.range(0..1).filter(|batch| batch.accounts()[0].0 == account)
142        else {
143            return Ok(Some(Vec::new()))
144        };
145
146        let mut requests = Vec::new();
147        for slot in repairs.slots(account).take(self.max_repair_slots) {
148            let request = GetStorageRangesMessage {
149                request_id: self.context.next_request_id(),
150                root_hash: batch.state_root(),
151                account_hashes: vec![account],
152                starting_hash: slot.into(),
153                limit_hash: slot.into(),
154                response_bytes: self.context.response_bytes(),
155            };
156            let downloader = StorageRangeDownloader::new(
157                self.context.client().clone(),
158                request,
159                &batch,
160                self.context.runtime().clone(),
161            )?;
162            requests.push(async move { downloader.await.map(|outcome| (slot, outcome)) });
163        }
164
165        let requested = requests.len();
166        let mut values = Vec::new();
167        for response in join_all(requests).await {
168            let (slot, StorageRangeOutcome::Verified(ranges)) = response? else { continue };
169            // Proved from the slot, so one keyed past it means the storage holds none there.
170            let value = ranges
171                .into_ranges()
172                .into_iter()
173                .next()
174                .and_then(|range| range.slots.first().copied())
175                .filter(|(key, _)| *key == slot)
176                .map_or(U256::ZERO, |(_, value)| value);
177            values.push((slot, value));
178        }
179        // An account scheduled without slots has nothing to wait for.
180        Ok((!values.is_empty() || requested == 0).then_some(values))
181    }
182}
183
184impl<C, F> fmt::Debug for StorageRangeDownload<C, F> {
185    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
186        f.debug_struct("StorageRangeDownload")
187            .field("context", &self.context)
188            .field("max_accounts", &self.max_accounts)
189            .field("max_repair_slots", &self.max_repair_slots)
190            .finish()
191    }
192}
193
194/// What one request of a [`StorageRangeDownload`] produced.
195#[derive(Debug)]
196pub enum StorageRangeStep {
197    /// A response was committed, reaching this progress.
198    Committed(StorageProgress),
199    /// The peer did not serve the storage, so the progress stays where it was.
200    Unavailable {
201        /// Peer that answered without the state, so the retry can go elsewhere.
202        peer_id: PeerId,
203    },
204    /// Every contract in the range has its storage persisted.
205    Complete,
206}
207
208// Splits a response into per-contract chunks. Only the last contract returned can be part way
209// through, and it continues where a follow-up request would resume.
210fn chunks(
211    ranges: VerifiedStorageRanges,
212    batch: VerifiedAccountBatch<'_>,
213    from: B256,
214) -> Result<Vec<StorageChunk>, SnapSyncError> {
215    let roots: Vec<B256> =
216        batch.accounts().iter().map(|(_, account)| account.storage_root).collect();
217    let resume = ranges.follow_up(0, batch)?.map(|(request, _)| {
218        (request.account_hashes[0], request.starting_hash.unwrap_or(B256::ZERO))
219    });
220    Ok(ranges
221        .into_ranges()
222        .into_iter()
223        .zip(roots)
224        .enumerate()
225        .map(|(index, (range, storage_root))| {
226            // Contracts after the first are requested whole.
227            let from = if index == 0 { from } else { B256::ZERO };
228            let next =
229                resume.filter(|(account, _)| *account == range.account_hash).map(|(_, slot)| slot);
230            StorageChunk::new(range.account_hash, storage_root, from, range.slots, next)
231        })
232        .collect())
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use crate::{
239        test_utils::{
240            account, generation, hashed_factory_with_roots, key, state_root, storage_ranges,
241            storage_root_of, stored_slots, verified_range, verified_repair, ScriptedSnapClient,
242        },
243        SnapAccountStore, SnapAttemptStore, SnapCatchUpStore, StateRepairs,
244    };
245    use alloy_eips::BlockNumHash;
246    use alloy_primitives::U256;
247    use reth_eth_wire_types::snap::StorageRangesMessage;
248    use reth_network_p2p::{error::PeerRequestResult, snap::client::SnapResponse};
249    use reth_network_peers::WithPeerId;
250    use reth_provider::{test_utils::MockNodeTypesWithDB, ProviderFactory};
251    use reth_trie_common::TrieAccount;
252    use std::sync::Arc;
253
254    const FAR: B256 = B256::repeat_byte(0xaa);
255
256    type Factory = ProviderFactory<MockNodeTypesWithDB>;
257    type Download = StorageRangeDownload<Arc<ScriptedSnapClient>, Factory>;
258
259    // Storage served over several responses.
260    fn large() -> Vec<(B256, U256)> {
261        vec![(key(1), U256::from(11)), (key(2), U256::from(12)), (key(3), U256::from(13))]
262    }
263
264    fn small() -> Vec<(B256, U256)> {
265        vec![(key(9), U256::from(19))]
266    }
267
268    fn contract(nonce: u64, slots: &[(B256, U256)]) -> TrieAccount {
269        let mut contract = account(nonce);
270        contract.storage_root = storage_root_of(slots);
271        contract
272    }
273
274    // A plain account, a large and a small contract, and a far plain account.
275    fn accounts() -> Vec<(B256, TrieAccount)> {
276        vec![
277            (key(1), account(1)),
278            (key(2), contract(2, &large())),
279            (key(3), contract(3, &small())),
280            (FAR, account(4)),
281        ]
282    }
283
284    // An attempt that fetched all of `accounts` as one range, not yet committed.
285    fn started(accounts: &[(B256, TrieAccount)]) -> (Factory, VerifiedRange) {
286        started_moving_to(accounts, accounts)
287    }
288
289    // `started`, with the block after the pivot committing to the state `moved`.
290    fn started_moving_to(
291        accounts: &[(B256, TrieAccount)],
292        moved: &[(B256, TrieAccount)],
293    ) -> (Factory, VerifiedRange) {
294        let factory = hashed_factory_with_roots(&[state_root(accounts), state_root(moved)]);
295        let provider = factory.database_provider_rw().unwrap();
296        let write = provider.start_snap_attempt(generation(1, state_root(accounts))).unwrap();
297        provider.start_account_coverage(write).unwrap();
298        provider.commit().unwrap();
299        let range = verified_range(accounts, 0..accounts.len(), B256::ZERO, &[]);
300        (factory, VerifiedRange::new(write, range))
301    }
302
303    fn download(
304        responses: impl IntoIterator<Item = PeerRequestResult<SnapResponse>>,
305        factory: Factory,
306    ) -> (Arc<ScriptedSnapClient>, Download) {
307        let client = Arc::new(ScriptedSnapClient::new(responses));
308        (Arc::clone(&client), StorageRangeDownload::new(client, factory, Runtime::test()))
309    }
310
311    async fn committed(download: &mut Download, range: &VerifiedRange) -> StorageProgress {
312        match download.next(range).await.unwrap() {
313            StorageRangeStep::Committed(progress) => progress,
314            step => panic!("expected a commit, got {step:?}"),
315        }
316    }
317
318    fn slots_of(factory: &Factory, account: B256) -> Vec<(B256, U256)> {
319        stored_slots(&factory.database_provider_ro().unwrap(), account)
320    }
321
322    #[tokio::test]
323    async fn a_contract_spanning_responses_is_committed_response_by_response() {
324        let accounts = accounts();
325        let (factory, range) = started(&accounts);
326        let (large, small) = (large(), small());
327        let responses = [
328            // The large contract's first slot; the proof shows more follow.
329            storage_ranges(1, &[&large[..1]], &large, &[B256::ZERO, key(1)]),
330            // The rest of it, without the small contract.
331            storage_ranges(2, &[&large[1..]], &large, &[key(2), key(3)]),
332            storage_ranges(3, &[&small[..]], &small, &[]),
333        ];
334        let (client, mut download) = download(responses, factory.clone());
335
336        let progress = committed(&mut download, &range).await;
337        assert_eq!(progress.resume_at(key(2)), Some(key(2)));
338        assert_eq!(slots_of(&factory, key(2)), large[..1]);
339
340        assert!(committed(&mut download, &range).await.is_complete(key(2)));
341        assert!(committed(&mut download, &range).await.is_complete(key(3)));
342        assert!(matches!(download.next(&range).await.unwrap(), StorageRangeStep::Complete));
343        assert_eq!(
344            *client.storage_requests(),
345            [
346                (vec![key(2), key(3)], B256::ZERO),
347                (vec![key(2), key(3)], key(2)),
348                (vec![key(3)], B256::ZERO),
349            ]
350        );
351
352        // The range commits against the persisted storage, none of it supplied in memory.
353        let provider = factory.database_provider_rw().unwrap();
354        let coverage = provider
355            .commit_account_range(range.write(), range.range(), Default::default(), Vec::new())
356            .unwrap();
357        provider.commit().unwrap();
358        assert!(coverage.is_complete());
359        assert_eq!(slots_of(&factory, key(2)), large);
360        assert_eq!(slots_of(&factory, key(3)), small);
361    }
362
363    #[tokio::test]
364    async fn an_unavailable_response_leaves_the_progress_in_place() {
365        let accounts = accounts();
366        let (factory, range) = started(&accounts);
367        let large = large();
368        let peer = PeerId::random();
369        let empty = StorageRangesMessage { request_id: 2, slots: Vec::new(), proof: Vec::new() };
370        let responses = [
371            storage_ranges(1, &[&large[..1]], &large, &[B256::ZERO, key(1)]),
372            Ok(WithPeerId::new(peer, SnapResponse::StorageRanges(empty))),
373            storage_ranges(3, &[&large[1..]], &large, &[key(2), key(3)]),
374        ];
375        let (client, mut download) = download(responses, factory.clone());
376        let progress = committed(&mut download, &range).await;
377
378        let step = download.next(&range).await.unwrap();
379
380        assert!(matches!(step, StorageRangeStep::Unavailable { peer_id } if peer_id == peer));
381        let provider = factory.database_provider_ro().unwrap();
382        assert_eq!(provider.storage_progress(range.write(), range.origin()).unwrap(), progress);
383        drop(provider);
384        assert!(committed(&mut download, &range).await.is_complete(key(2)));
385        let origins: Vec<_> = client.storage_requests().iter().map(|(_, from)| *from).collect();
386        assert_eq!(origins, [B256::ZERO, key(2), key(2)]);
387    }
388
389    #[tokio::test]
390    async fn requests_ask_for_at_most_the_configured_contracts() {
391        let accounts = accounts();
392        let (factory, range) = started(&accounts);
393        let (large, small) = (large(), small());
394        let responses = [
395            storage_ranges(1, &[&large[..]], &large, &[]),
396            storage_ranges(2, &[&small[..]], &small, &[]),
397        ];
398        let (client, download) = download(responses, factory);
399        let mut download = download.with_max_accounts(1);
400
401        committed(&mut download, &range).await;
402        committed(&mut download, &range).await;
403
404        assert!(matches!(download.next(&range).await.unwrap(), StorageRangeStep::Complete));
405        assert_eq!(
406            *client.storage_requests(),
407            [(vec![key(2)], B256::ZERO), (vec![key(3)], B256::ZERO)]
408        );
409    }
410
411    #[tokio::test]
412    async fn a_range_without_contracts_requests_no_storage() {
413        let accounts = vec![(key(1), account(1)), (FAR, account(2))];
414        let (factory, range) = started(&accounts);
415        let (client, mut download) = download([], factory);
416
417        assert!(matches!(download.next(&range).await.unwrap(), StorageRangeStep::Complete));
418        assert!(client.storage_requests().is_empty());
419    }
420
421    #[tokio::test]
422    async fn storage_fetched_before_the_pivot_moved_is_not_resumed() {
423        let accounts = accounts();
424        let (factory, range) = started(&accounts);
425        let large = large();
426        let first = storage_ranges(1, &[&large[..1]], &large, &[B256::ZERO, key(1)]);
427        let (client, mut download) = download([first], factory.clone());
428        committed(&mut download, &range).await;
429        let provider = factory.database_provider_rw().unwrap();
430        provider.advance_snap_pivot(range.write(), generation(2, state_root(&accounts))).unwrap();
431        provider.commit().unwrap();
432
433        assert!(matches!(download.next(&range).await, Err(SnapSyncError::StaleWrite { .. })));
434        assert_eq!(client.storage_requests().len(), 1);
435    }
436
437    #[tokio::test]
438    async fn storage_part_way_through_resumes_at_the_new_root() {
439        let accounts = accounts();
440        let (large, small) = (large(), small());
441        // The new pivot turned the first account into a contract, ahead of the carried one.
442        let mut moved = accounts.clone();
443        moved[0].1 = contract(1, &small);
444        let (factory, range) = started_moving_to(&accounts, &moved);
445        let responses = [
446            storage_ranges(1, &[&large[..1]], &large, &[B256::ZERO, key(1)]),
447            storage_ranges(2, &[&small[..]], &small, &[]),
448            storage_ranges(3, &[&large[1..]], &large, &[key(2), key(3)]),
449            storage_ranges(4, &[&small[..]], &small, &[]),
450        ];
451        let (client, mut download) = download(responses, factory.clone());
452        committed(&mut download, &range).await;
453
454        let provider = factory.database_provider_rw().unwrap();
455        let write =
456            provider.advance_snap_pivot(range.write(), generation(2, state_root(&moved))).unwrap();
457        provider.commit().unwrap();
458        let range =
459            VerifiedRange::new(write, verified_range(&moved, 0..moved.len(), B256::ZERO, &[]));
460        for _ in 0..3 {
461            committed(&mut download, &range).await;
462        }
463        assert!(matches!(download.next(&range).await.unwrap(), StorageRangeStep::Complete));
464        assert_eq!(
465            *client.storage_requests(),
466            [
467                (vec![key(2), key(3)], B256::ZERO),
468                (vec![key(1)], B256::ZERO),
469                (vec![key(2), key(3)], key(2)),
470                (vec![key(3)], B256::ZERO),
471            ]
472        );
473
474        // The carried slot holds pivot 1's value until the lists reach the new pivot.
475        let provider = factory.database_provider_rw().unwrap();
476        assert!(matches!(
477            provider.commit_account_range(write, range.range(), Default::default(), Vec::new()),
478            Err(SnapSyncError::CatchUpBehindPivot { applied: 1, pivot: 2 })
479        ));
480        let block = BlockNumHash::new(2, B256::repeat_byte(2));
481        provider.commit_block_access_list(write, block, B256::repeat_byte(1), &[]).unwrap();
482        let coverage = provider
483            .commit_account_range(write, range.range(), Default::default(), Vec::new())
484            .unwrap();
485        provider.commit().unwrap();
486        assert!(coverage.is_complete());
487        assert_eq!(slots_of(&factory, key(1)), small);
488        assert_eq!(slots_of(&factory, key(2)), large);
489    }
490
491    #[tokio::test]
492    async fn a_page_ending_before_a_carried_contract_hands_it_to_the_next_range() {
493        let accounts = accounts();
494        let (factory, range) = started(&accounts);
495        let large = large();
496        let first = storage_ranges(1, &[&large[..1]], &large, &[B256::ZERO, key(1)]);
497        let (_, mut download) = download([first], factory.clone());
498        committed(&mut download, &range).await;
499        let provider = factory.database_provider_rw().unwrap();
500        let write = provider
501            .advance_snap_pivot(range.write(), generation(2, state_root(&accounts)))
502            .unwrap();
503        let block = BlockNumHash::new(2, B256::repeat_byte(2));
504        provider.commit_block_access_list(write, block, B256::repeat_byte(1), &[]).unwrap();
505
506        // The new root's page stops before the contract part way through.
507        let page = verified_range(&accounts, 0..1, B256::ZERO, &[B256::ZERO, key(1)]);
508        let next = provider
509            .commit_account_range(write, &page, Default::default(), Vec::new())
510            .unwrap()
511            .next()
512            .unwrap();
513
514        assert!(next <= key(2));
515        assert_eq!(provider.storage_progress(write, next).unwrap().resume_at(key(2)), Some(key(2)));
516    }
517
518    // `accounts`' large contract fetched on its own, with `slots` scheduled for repair.
519    fn repairing(accounts: &[(B256, TrieAccount)], slots: &[B256]) -> (Factory, VerifiedRange) {
520        let (factory, _) = started(accounts);
521        let provider = factory.database_provider_rw().unwrap();
522        let write = provider.active_snap_write().unwrap().unwrap();
523        let mut repairs = StateRepairs::default();
524        for slot in slots {
525            repairs.insert_slot(key(2), *slot);
526        }
527        provider.schedule_snap_repairs(write, repairs).unwrap();
528        provider.commit().unwrap();
529        let range = verified_repair(accounts, 1..2, key(2), &[key(2)]);
530        (factory, VerifiedRange::new(write, range))
531    }
532
533    #[tokio::test]
534    async fn repair_slots_read_zero_where_the_pivot_holds_none() {
535        let accounts = accounts();
536        let (factory, range) = repairing(&accounts, &[key(1), key(5)]);
537        let large = large();
538        let responses = [
539            storage_ranges(1, &[&large[..1]], &large, &[key(1)]),
540            // Nothing is stored at key 5, which the proof shows.
541            storage_ranges(2, &[&[]], &large, &[key(5)]),
542        ];
543        let (client, mut download) = download(responses, factory);
544
545        let values = download.repair_slots(&range).await.unwrap();
546
547        assert_eq!(values, Some(vec![(key(1), U256::from(11)), (key(5), U256::ZERO)]));
548        assert_eq!(*client.storage_requests(), [(vec![key(2)], key(1)), (vec![key(2)], key(5))]);
549    }
550
551    #[tokio::test]
552    async fn a_contract_scheduled_without_slots_fetches_none() {
553        let accounts = accounts();
554        let (factory, range) = repairing(&accounts, &[]);
555        let provider = factory.database_provider_rw().unwrap();
556        let mut repairs = StateRepairs::default();
557        repairs.insert_account(key(2));
558        provider.schedule_snap_repairs(range.write(), repairs).unwrap();
559        provider.commit().unwrap();
560        let (client, mut download) = download([], factory);
561
562        assert_eq!(download.repair_slots(&range).await.unwrap(), Some(Vec::new()));
563        assert!(client.storage_requests().is_empty());
564    }
565
566    #[tokio::test]
567    async fn an_unserved_repair_slot_fetches_nothing() {
568        let accounts = accounts();
569        let (factory, range) = repairing(&accounts, &[key(1)]);
570        let (_, mut download) = download([storage_ranges(1, &[], &[], &[])], factory);
571
572        assert_eq!(download.repair_slots(&range).await.unwrap(), None);
573    }
574
575    #[tokio::test]
576    async fn served_repair_slots_survive_an_unserved_one() {
577        let accounts = accounts();
578        let (factory, range) = repairing(&accounts, &[key(1), key(5)]);
579        let large = large();
580        let responses = [
581            storage_ranges(1, &[&large[..1]], &large, &[key(1)]),
582            storage_ranges(2, &[], &[], &[]),
583        ];
584        let (_, mut download) = download(responses, factory);
585
586        let values = download.repair_slots(&range).await.unwrap();
587
588        // Key 5 stays scheduled for the next batch.
589        assert_eq!(values, Some(vec![(key(1), U256::from(11))]));
590    }
591
592    #[tokio::test]
593    async fn repair_slots_are_fetched_in_bounded_batches() {
594        let accounts = accounts();
595        let (factory, range) = repairing(&accounts, &[key(1), key(5)]);
596        let large = large();
597        let responses = [storage_ranges(1, &[&large[..1]], &large, &[key(1)])];
598        let (client, download) = download(responses, factory);
599        let mut download = download.with_max_repair_slots(1);
600
601        let values = download.repair_slots(&range).await.unwrap();
602
603        // Key 5 waits for the next batch, after the account commits this one.
604        assert_eq!(values, Some(vec![(key(1), U256::from(11))]));
605        assert_eq!(*client.storage_requests(), [(vec![key(2)], key(1))]);
606    }
607}
reth_snap_sync/storage/download.rs

reth_snap_sync/storage/
download.rs
