reth_snap_sync/catch_up/
download.rs

1//! Requests the lists of the canonical blocks between the applied one and the catch-up target.
2//!
3//! Headers are read again each request, so a reorged anchor is caught before peers are asked.
4
5use crate::{
6    common::DownloadContext, BlockWithAccessListHash, CatchUpProgress, SnapAttemptStore,
7    SnapCatchUpStore, SnapSyncError, SnapWrite,
8};
9use alloy_eips::eip7928::bal::DecodedBal;
10use reth_db_api::transaction::DbTxMut;
11use reth_downloaders::snap::{BlockAccessListDownloader, BlockAccessListOutcome};
12use reth_eth_wire_types::snap::GetBlockAccessListsMessage;
13use reth_network_p2p::snap::client::SnapClient;
14use reth_network_peers::PeerId;
15use reth_primitives_traits::{AlloyBlockHeader, SealedHeader};
16use reth_storage_api::{
17    BlockHashReader, DBProvider, DatabaseProviderFactory, HeaderProvider, MetadataProvider,
18    MetadataWriter, StateWriter,
19};
20use reth_tasks::Runtime;
21use std::fmt;
22
23/// Default soft response limit for block access list requests, as EIP-8189 recommends.
24pub const DEFAULT_BAL_RESPONSE_BYTES: u64 = 2 * 1024 * 1024;
25
26/// Default number of blocks asked for per request, chosen so that many average 60M gas lists
27/// still fit under [`DEFAULT_BAL_RESPONSE_BYTES`].
28pub const DEFAULT_CATCH_UP_BLOCKS: u64 = 28;
29
30/// Applies the lists of the canonical blocks after the applied one, one request at a time.
31pub struct BlockAccessListCatchUp<C, F> {
32    context: DownloadContext<C, F>,
33    // Blocks asked for per request.
34    max_blocks: u64,
35}
36
37impl<C, F> BlockAccessListCatchUp<C, F> {
38    /// Creates a catch-up that requests the lists the applied state does not carry yet.
39    pub const fn new(client: C, factory: F, runtime: Runtime) -> Self {
40        let mut context = DownloadContext::new(client, factory, runtime);
41        context.set_response_bytes(DEFAULT_BAL_RESPONSE_BYTES);
42        Self { context, max_blocks: DEFAULT_CATCH_UP_BLOCKS }
43    }
44
45    /// Returns this catch-up asking peers for at most `response_bytes` per response.
46    pub const fn with_response_bytes(mut self, response_bytes: u64) -> Self {
47        self.context.set_response_bytes(response_bytes);
48        self
49    }
50
51    /// Returns this catch-up asking for at most `max_blocks` blocks per request, at least one.
52    pub const fn with_max_blocks(mut self, max_blocks: u64) -> Self {
53        self.max_blocks = if max_blocks == 0 { 1 } else { max_blocks };
54        self
55    }
56}
57
58impl<C, F> BlockAccessListCatchUp<C, F>
59where
60    C: SnapClient + Clone + Unpin,
61    F: DatabaseProviderFactory + Clone + 'static,
62    F::Provider: HeaderProvider + MetadataProvider + BlockHashReader,
63    F::ProviderRW:
64        BlockHashReader + MetadataProvider + MetadataWriter + StateWriter + DBProvider<Tx: DbTxMut>,
65{
66    /// Requests the lists of the canonical blocks between the applied one and `target`, and
67    /// commits those continuing it, in order.
68    ///
69    /// A list a peer leaves out ends the run this call commits, leaving it and every block after
70    /// it to the next one.
71    pub async fn next(
72        &mut self,
73        write: SnapWrite,
74        target: u64,
75    ) -> Result<CatchUpStep, SnapSyncError> {
76        let max_blocks = self.max_blocks;
77        let headers = self
78            .context
79            .read(move |provider| pending_headers(provider, write, target, max_blocks))
80            .await?;
81        if headers.is_empty() {
82            return Ok(CatchUpStep::Complete)
83        }
84
85        let requested = headers.iter().map(BlockWithAccessListHash::from).collect::<Vec<_>>();
86        let verified = match self.request(&requested).await? {
87            BlockAccessListOutcome::Verified(verified) => verified,
88            BlockAccessListOutcome::Unavailable { peer_id } => {
89                return Ok(CatchUpStep::Unavailable { peer_id })
90            }
91        };
92
93        let peer_id = verified.peer_id();
94        // A list applies only once every earlier block's has, so the run stops at the first block
95        // the peer left out.
96        let applied = verified
97            .into_block_access_lists()
98            .into_iter()
99            .zip(&requested)
100            .map_while(|((_, list), kept)| list.map(|list| (kept.block(), list)))
101            .collect::<Vec<_>>();
102        if applied.is_empty() {
103            return Ok(CatchUpStep::Unavailable { peer_id })
104        }
105
106        let blocks = applied.len();
107        let progress = self
108            .context
109            .commit(move |provider| {
110                let mut progress = None;
111                for (block, list) in applied {
112                    progress = Some(provider.commit_block_access_list(
113                        write,
114                        block.block,
115                        block.parent,
116                        list.as_bal(),
117                    )?);
118                }
119                Ok(progress.expect("the applied run is not empty"))
120            })
121            .await?;
122        Ok(CatchUpStep::Applied { progress, blocks })
123    }
124
125    /// Fetches the lists of `blocks`, which a reorg orphaned, in order.
126    ///
127    /// A response cut short at the byte limit keeps the lists it served and the rest is asked for
128    /// again. `Ok(None)` once a request serves none of the remaining lists, since repairing the
129    /// state needs every one.
130    pub async fn orphaned_lists(
131        &mut self,
132        blocks: &[BlockWithAccessListHash],
133    ) -> Result<Option<Vec<DecodedBal>>, SnapSyncError> {
134        let mut lists = Vec::with_capacity(blocks.len());
135        while lists.len() < blocks.len() {
136            let end = blocks.len().min(lists.len() + self.max_blocks as usize);
137            let BlockAccessListOutcome::Verified(verified) =
138                self.request(&blocks[lists.len()..end]).await?
139            else {
140                return Ok(None)
141            };
142            let served = lists.len();
143            lists
144                .extend(verified.into_block_access_lists().into_iter().map_while(|(_, list)| list));
145            if lists.len() == served {
146                return Ok(None)
147            }
148        }
149        Ok(Some(lists))
150    }
151
152    // Requests the lists of `blocks`, each authenticated against its block access list hash.
153    async fn request(
154        &mut self,
155        blocks: &[BlockWithAccessListHash],
156    ) -> Result<BlockAccessListOutcome, SnapSyncError> {
157        let request = GetBlockAccessListsMessage {
158            request_id: self.context.next_request_id(),
159            block_hashes: blocks.iter().map(|kept| kept.block().block.hash).collect(),
160            response_bytes: self.context.response_bytes(),
161        };
162        let downloader = BlockAccessListDownloader::with_access_list_hashes(
163            self.context.client().clone(),
164            request,
165            blocks.iter().map(BlockWithAccessListHash::block_access_list_hash).collect(),
166            self.context.runtime().clone(),
167        )?;
168        Ok(downloader.await?)
169    }
170}
171
172impl<C, F> fmt::Debug for BlockAccessListCatchUp<C, F> {
173    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
174        f.debug_struct("BlockAccessListCatchUp")
175            .field("context", &self.context)
176            .field("max_blocks", &self.max_blocks)
177            .finish()
178    }
179}
180
181/// What one request of a [`BlockAccessListCatchUp`] produced.
182#[derive(Debug)]
183pub enum CatchUpStep {
184    /// Lists were committed, carrying the downloaded state this far.
185    Applied {
186        /// How far the state is carried now.
187        progress: CatchUpProgress,
188        /// Blocks this request carried it past.
189        blocks: usize,
190    },
191    /// The peer held no list for the next block, so the state stays where it is.
192    Unavailable {
193        /// Peer that answered without it, so the retry can go elsewhere.
194        peer_id: PeerId,
195    },
196    /// Every block through the target is applied.
197    Complete,
198}
199
200// The canonical headers continuing the applied block, at most `max_blocks` of them and none past
201// `target` or the pivot.
202//
203// Both anchors must remain canonical: downloaded ranges can contain changes from the pivot
204// even when catch-up has only reached an earlier block.
205fn pending_headers<P: HeaderProvider + MetadataProvider + BlockHashReader>(
206    provider: &P,
207    write: SnapWrite,
208    target: u64,
209    max_blocks: u64,
210) -> Result<Vec<SealedHeader<P::Header>>, SnapSyncError> {
211    let progress = provider.catch_up_progress(write)?.ok_or(SnapSyncError::NoCatchUpProgress)?;
212    let applied = progress.applied();
213    let canonical = provider
214        .sealed_header(applied.number)?
215        .ok_or(SnapSyncError::MissingHeader { block: applied.number })?;
216    if canonical.hash() != applied.hash {
217        return Err(SnapSyncError::ForkedBlock { expected: applied.hash, got: canonical.hash() })
218    }
219    // Pending ranges are proved against the pivot, so no list past it applies.
220    let target = target.min(provider.authorize_canonical_snap_write(write)?.pivot().number);
221    if target <= applied.number {
222        return Ok(Vec::new())
223    }
224
225    let end = target.min(applied.number.saturating_add(max_blocks));
226    let headers = provider.sealed_headers_range(progress.next()..=end)?;
227    if headers.is_empty() {
228        return Err(SnapSyncError::MissingHeader { block: progress.next() })
229    }
230    // A list is only authenticated by the header of the block it belongs to, so a gap in the
231    // headers fails the request, unlike a gap in the lists, which only ends the run.
232    for (number, header) in (progress.next()..=end).zip(&headers) {
233        if header.number() != number {
234            return Err(SnapSyncError::MissingHeader { block: number })
235        }
236    }
237    Ok(headers)
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243    use crate::{
244        test_utils::{
245            account, hashed_factory, key, state_root, verified_range, BalChain, ScriptedSnapClient,
246            SnapStateSnapshot,
247        },
248        SnapAccountStore, SnapGeneration,
249    };
250    use alloy_eip7928::{AccountChanges, BalanceChange, BlockAccessIndex};
251    use alloy_primitives::{keccak256, Address, B256, U256};
252    use reth_db_api::{cursor::DbCursorRO, tables, transaction::DbTx};
253    use reth_network_p2p::{error::PeerRequestResult, snap::client::SnapResponse};
254    use reth_provider::{
255        test_utils::{insert_headers, MockNodeTypesWithDB},
256        ProviderFactory,
257    };
258    use reth_trie_common::{TrieAccount, EMPTY_ROOT_HASH};
259    use std::sync::Arc;
260
261    type Factory = ProviderFactory<MockNodeTypesWithDB>;
262    type CatchUp = BlockAccessListCatchUp<Arc<ScriptedSnapClient>, Factory>;
263
264    // The account every fixture list credits.
265    const CHANGED: Address = Address::repeat_byte(0xaa);
266    // Pivot the fixture chain is anchored to.
267    const PIVOT: u64 = 2;
268
269    fn accounts() -> Vec<(B256, TrieAccount)> {
270        let mut accounts = vec![(key(1), account(1)), (keccak256(CHANGED), account(2))];
271        accounts.sort_by_key(|(hashed_address, _)| *hashed_address);
272        accounts
273    }
274
275    // A list crediting `CHANGED` with `balance`.
276    fn credit(balance: u64) -> Vec<AccountChanges> {
277        vec![AccountChanges::new(CHANGED)
278            .with_balance_change(BalanceChange::new(BlockAccessIndex::new(1), U256::from(balance)))]
279    }
280
281    // Three blocks after the pivot, crediting the account 10, 20 and 30.
282    fn chain() -> BalChain {
283        BalChain::new(PIVOT, [credit(10), credit(20), credit(30)])
284    }
285
286    // An attempt started at `chain`'s pivot, with every account downloaded and nothing applied,
287    // then moved to the chain's last block.
288    fn started(chain: &BalChain, accounts: &[(B256, TrieAccount)]) -> (Factory, SnapWrite) {
289        let factory = hashed_factory();
290        insert_headers(&factory, &chain.headers_with_root(state_root(accounts)));
291        let provider = factory.database_provider_rw().unwrap();
292        let write = provider.start_snap_attempt(chain.generation(state_root(accounts))).unwrap();
293        provider.start_account_coverage(write).unwrap();
294        let range = verified_range(accounts, 0..accounts.len(), B256::ZERO, &[]);
295        provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
296        let tip = SnapGeneration::new(chain.tip(), state_root(accounts));
297        let write = provider.advance_snap_pivot(write, tip).unwrap();
298        provider.commit().unwrap();
299        (factory, write)
300    }
301
302    // The reorg replaces the pivot alone: the applied block and the one the lists carry the state
303    // through stay canonical, so only the pivot check can refuse what follows. Ranges downloaded
304    // at that pivot hold a balance change the replacement branch never made.
305    fn pivot_reorg_fixture() -> (Factory, SnapWrite, BalChain, BalChain) {
306        let chain = BalChain::new(PIVOT, [Vec::new(), credit(30)]);
307        let replacement = BalChain::new(PIVOT, [Vec::new(), Vec::new()]);
308        assert_eq!(chain.block(1), replacement.block(1));
309        assert_ne!(chain.tip(), replacement.tip());
310        let mut accounts = accounts();
311        let factory = hashed_factory();
312        let mut current = accounts.clone();
313        current.iter_mut().find(|(key, _)| *key == keccak256(CHANGED)).unwrap().1.balance =
314            U256::from(30);
315        insert_headers(
316            &factory,
317            &chain.headers_with_roots(&[
318                (PIVOT, state_root(&accounts)),
319                (chain.tip().number, state_root(&current)),
320            ]),
321        );
322        let provider = factory.database_provider_rw().unwrap();
323        let write = provider.start_snap_attempt(chain.generation(state_root(&accounts))).unwrap();
324        provider.start_account_coverage(write).unwrap();
325        accounts.iter_mut().find(|(key, _)| *key == keccak256(CHANGED)).unwrap().1.balance =
326            U256::from(30);
327        let write = provider
328            .advance_snap_pivot(write, SnapGeneration::new(chain.tip(), state_root(&accounts)))
329            .unwrap();
330        let range = verified_range(&accounts, 0..accounts.len(), B256::ZERO, &[]);
331        provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
332        provider.commit().unwrap();
333        (factory, write, chain, replacement)
334    }
335
336    fn catch_up(
337        responses: impl IntoIterator<Item = PeerRequestResult<SnapResponse>>,
338        factory: Factory,
339    ) -> (Arc<ScriptedSnapClient>, CatchUp) {
340        catch_up_with(responses, factory, DEFAULT_CATCH_UP_BLOCKS)
341    }
342
343    fn catch_up_with(
344        responses: impl IntoIterator<Item = PeerRequestResult<SnapResponse>>,
345        factory: Factory,
346        max_blocks: u64,
347    ) -> (Arc<ScriptedSnapClient>, CatchUp) {
348        let client = Arc::new(ScriptedSnapClient::new(responses));
349        let catch_up = BlockAccessListCatchUp::new(Arc::clone(&client), factory, Runtime::test());
350        (client, catch_up.with_max_blocks(max_blocks))
351    }
352
353    // Blocks of `chain` after the pivot, kept as a reorg orphaning it finds them, with their
354    // hashes.
355    fn orphaned(chain: &BalChain) -> (Vec<BlockWithAccessListHash>, Vec<B256>) {
356        let headers = &chain.headers[PIVOT as usize + 1..];
357        (
358            headers.iter().map(BlockWithAccessListHash::from).collect(),
359            headers.iter().map(SealedHeader::hash).collect(),
360        )
361    }
362
363    async fn applied(catch_up: &mut CatchUp, write: SnapWrite, target: u64) -> CatchUpProgress {
364        match catch_up.next(write, target).await.unwrap() {
365            CatchUpStep::Applied { progress, .. } => progress,
366            step => panic!("expected an application, got {step:?}"),
367        }
368    }
369
370    fn balance(factory: &Factory) -> U256 {
371        let provider = factory.database_provider_ro().unwrap();
372        provider
373            .tx_ref()
374            .get::<tables::HashedAccounts>(keccak256(CHANGED))
375            .unwrap()
376            .unwrap()
377            .balance
378    }
379
380    // The fixture accounts with `CHANGED` holding `balance`, plus `extra`.
381    fn accounts_with(
382        balance: u64,
383        extra: impl IntoIterator<Item = (B256, TrieAccount)>,
384    ) -> Vec<(B256, TrieAccount)> {
385        let mut accounts: Vec<_> = accounts().into_iter().chain(extra).collect();
386        accounts.iter_mut().find(|(key, _)| *key == keccak256(CHANGED)).unwrap().1.balance =
387            U256::from(balance);
388        accounts.sort_by_key(|(hashed_address, _)| *hashed_address);
389        accounts
390    }
391
392    // Root of the account trie the downloaded state holds, none of the fixture accounts having
393    // storage.
394    fn downloaded_root(factory: &Factory) -> B256 {
395        let provider = factory.database_provider_ro().unwrap();
396        let mut cursor = provider.tx_ref().cursor_read::<tables::HashedAccounts>().unwrap();
397        let accounts: Vec<_> = cursor
398            .walk(None)
399            .unwrap()
400            .map(|entry| {
401                let (hashed_address, account) = entry.unwrap();
402                (hashed_address, account.into_trie_account(EMPTY_ROOT_HASH))
403            })
404            .collect();
405        state_root(&accounts)
406    }
407
408    // An attempt at `chain`'s pivot holding `accounts[..served]`, proved against `accounts`.
409    fn partially_downloaded(
410        chain: &BalChain,
411        accounts: &[(B256, TrieAccount)],
412        served: usize,
413    ) -> (Factory, SnapWrite) {
414        let factory = hashed_factory();
415        insert_headers(&factory, &chain.headers_with_roots(&[(PIVOT, state_root(accounts))]));
416        let provider = factory.database_provider_rw().unwrap();
417        let write = provider.start_snap_attempt(chain.generation(state_root(accounts))).unwrap();
418        provider.start_account_coverage(write).unwrap();
419        let range =
420            verified_range(accounts, 0..served, B256::ZERO, &[B256::ZERO, accounts[served - 1].0]);
421        provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
422        provider.commit().unwrap();
423        (factory, write)
424    }
425
426    // Moves the attempt to the block `nth` after the chain's pivot, whose state is `accounts`.
427    fn advance(
428        factory: &Factory,
429        write: SnapWrite,
430        chain: &BalChain,
431        nth: usize,
432        accounts: &[(B256, TrieAccount)],
433    ) -> SnapWrite {
434        let provider = factory.database_provider_rw().unwrap();
435        let generation = SnapGeneration::new(chain.block(nth), state_root(accounts));
436        let write = provider.advance_snap_pivot(write, generation).unwrap();
437        provider.commit().unwrap();
438        write
439    }
440
441    // Commits the rest of the trie from the coverage cursor, proved against `accounts`.
442    fn download_rest(factory: &Factory, write: SnapWrite, accounts: &[(B256, TrieAccount)]) {
443        let provider = factory.database_provider_rw().unwrap();
444        let origin = provider.account_coverage(write).unwrap().unwrap().next().unwrap();
445        let served = accounts.iter().position(|(key, _)| *key >= origin).unwrap()..accounts.len();
446        let range = verified_range(accounts, served, origin, &[origin, accounts.last().unwrap().0]);
447        provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
448        provider.commit().unwrap();
449    }
450
451    #[tokio::test]
452    async fn every_list_a_response_carries_is_applied_in_block_order() {
453        let chain = chain();
454        let (factory, write) = started(&chain, &accounts());
455        let (client, mut catch_up) =
456            catch_up([chain.response(1, [Some(1), Some(2), Some(3)])], factory.clone());
457
458        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
459
460        assert_eq!(progress.applied(), chain.block(3));
461        // The last block's credit is the one that survives.
462        assert_eq!(balance(&factory), U256::from(30));
463        assert_eq!(
464            *client.block_requests(),
465            [(1..=3).map(|nth| chain.block(nth).hash).collect::<Vec<_>>()]
466        );
467        assert!(matches!(catch_up.next(write, PIVOT + 3).await.unwrap(), CatchUpStep::Complete));
468    }
469
470    #[tokio::test]
471    async fn a_list_a_peer_leaves_out_holds_back_the_blocks_after_it() {
472        let chain = chain();
473        let (factory, write) = started(&chain, &accounts());
474        let (client, mut catch_up) = catch_up(
475            [chain.response(1, [Some(1), None, Some(3)]), chain.response(2, [Some(2), Some(3)])],
476            factory.clone(),
477        );
478
479        // Only the block before the gap applies, although a later list was authenticated.
480        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
481        assert_eq!(progress.applied(), chain.block(1));
482        assert_eq!(balance(&factory), U256::from(10));
483
484        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
485        assert_eq!(progress.applied(), chain.block(3));
486        assert_eq!(balance(&factory), U256::from(30));
487        // The gap is asked for again, and nothing before it is.
488        assert_eq!(client.block_requests()[1], [chain.block(2).hash, chain.block(3).hash]);
489    }
490
491    #[tokio::test]
492    async fn a_response_cut_short_leaves_the_rest_pending() {
493        let chain = chain();
494        let (factory, write) = started(&chain, &accounts());
495        // A peer that stops at its soft byte limit answers fewer blocks than it was asked for.
496        let (client, mut catch_up) = catch_up(
497            [chain.response(1, [Some(1), Some(2)]), chain.response(2, [Some(3)])],
498            factory.clone(),
499        );
500
501        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
502        assert_eq!(progress.applied(), chain.block(2));
503
504        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
505        assert_eq!(progress.applied(), chain.block(3));
506        assert_eq!(balance(&factory), U256::from(30));
507        // Only the suffix the response left off is asked for again.
508        assert_eq!(client.block_requests()[1], [chain.block(3).hash]);
509    }
510
511    #[tokio::test]
512    async fn a_peer_holding_no_list_for_the_next_block_applies_nothing() {
513        let chain = chain();
514        let (factory, write) = started(&chain, &accounts());
515        let (_, mut catch_up) = catch_up([chain.response(1, [None, None, None])], factory.clone());
516
517        let step = catch_up.next(write, PIVOT + 3).await.unwrap();
518
519        assert!(matches!(step, CatchUpStep::Unavailable { .. }));
520        assert_eq!(balance(&factory), U256::ONE);
521    }
522
523    #[tokio::test]
524    async fn a_block_that_changes_nothing_is_still_carried_past() {
525        let chain = BalChain::new(PIVOT, [Vec::new()]);
526        let (factory, write) = started(&chain, &accounts());
527        let (_, mut catch_up) = catch_up([chain.response(1, [Some(1)])], factory.clone());
528
529        let progress = applied(&mut catch_up, write, PIVOT + 1).await;
530
531        assert_eq!(progress.applied(), chain.block(1));
532        assert_eq!(balance(&factory), U256::ONE);
533    }
534
535    #[tokio::test]
536    async fn only_the_blocks_a_request_can_carry_are_asked_for() {
537        let chain = chain();
538        let (factory, write) = started(&chain, &accounts());
539        let (client, mut catch_up) = catch_up_with([chain.response(1, [Some(1)])], factory, 1);
540
541        applied(&mut catch_up, write, PIVOT + 3).await;
542
543        assert_eq!(*client.block_requests(), [vec![chain.block(1).hash]]);
544    }
545
546    #[tokio::test]
547    async fn an_unbounded_request_asks_for_every_block_through_the_target() {
548        let chain = chain();
549        let (factory, write) = started(&chain, &accounts());
550        let (client, mut catch_up) =
551            catch_up_with([chain.response(1, [Some(1), Some(2), Some(3)])], factory, u64::MAX);
552
553        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
554
555        assert_eq!(progress.applied(), chain.block(3));
556        assert_eq!(
557            *client.block_requests(),
558            [(1..=3).map(|nth| chain.block(nth).hash).collect::<Vec<_>>()]
559        );
560    }
561
562    #[tokio::test]
563    async fn a_target_past_the_pivot_is_held_at_it() {
564        let chain = chain();
565        let factory = hashed_factory();
566        insert_headers(&factory, &chain.headers_with_root(state_root(&accounts())));
567        let provider = factory.database_provider_rw().unwrap();
568        let write = provider.start_snap_attempt(chain.generation(state_root(&accounts()))).unwrap();
569        provider.commit().unwrap();
570        let (client, mut catch_up) = catch_up([], factory);
571
572        assert!(matches!(catch_up.next(write, PIVOT + 3).await.unwrap(), CatchUpStep::Complete));
573        assert!(client.block_requests().is_empty());
574    }
575
576    #[tokio::test]
577    async fn an_applied_block_the_canonical_chain_no_longer_holds_is_reported() {
578        let chain = chain();
579        let accounts = accounts();
580        // The chain the node holds instead, forking before the pivot as a reorg leaves it.
581        let reorged = BalChain::new(0, [credit(1), credit(2), credit(3)]);
582        let factory = hashed_factory();
583        insert_headers(&factory, &chain.headers_with_root(state_root(&accounts)));
584        let provider = factory.database_provider_rw().unwrap();
585        let write = provider.start_snap_attempt(chain.generation(state_root(&accounts))).unwrap();
586        provider.start_account_coverage(write).unwrap();
587        provider.commit().unwrap();
588        reorged.replace_after(&factory, 0);
589        let (_, mut catch_up) = catch_up([], factory);
590
591        let forked = catch_up.next(write, PIVOT + 3).await;
592
593        assert!(matches!(forked, Err(SnapSyncError::ForkedBlock { .. })));
594    }
595
596    #[tokio::test]
597    async fn a_pivot_the_canonical_chain_no_longer_holds_is_reported() {
598        let (factory, write, chain, replacement) = pivot_reorg_fixture();
599        let before = SnapStateSnapshot::read(&factory.database_provider_ro().unwrap());
600        replacement.replace_tip(&factory);
601        let (client, mut catch_up) = catch_up([], factory.clone());
602
603        // Even an already-applied target must reject the orphan instead of reporting completion.
604        for target in [PIVOT, PIVOT + 2] {
605            assert!(matches!(
606                catch_up.next(write, target).await,
607                Err(SnapSyncError::NonCanonicalBlock { block, hash })
608                    if block == PIVOT + 2 && hash == chain.tip().hash
609            ));
610        }
611        assert!(client.block_requests().is_empty());
612        let provider = factory.database_provider_ro().unwrap();
613        // The block whose list would apply next is untouched by the reorg.
614        assert_eq!(provider.block_hash(PIVOT + 1).unwrap(), Some(chain.block(1).hash));
615        assert_eq!(provider.block_hash(PIVOT).unwrap(), Some(chain.block(0).hash));
616        assert_eq!(provider.catch_up_progress(write).unwrap().unwrap().applied(), chain.block(0));
617        assert_eq!(SnapStateSnapshot::read(&provider), before);
618    }
619
620    #[tokio::test]
621    async fn a_pivot_reorg_during_a_bal_request_changes_nothing() {
622        let (factory, write, chain, replacement) = pivot_reorg_fixture();
623        let before = SnapStateSnapshot::read(&factory.database_provider_ro().unwrap());
624        let reorg_factory = factory.clone();
625        let replacement_hash = replacement.tip().hash;
626        let client = Arc::new(
627            ScriptedSnapClient::new([chain.response(1, [Some(1)])])
628                .on_block_request(move || replacement.replace_tip(&reorg_factory)),
629        );
630        let mut catch_up =
631            BlockAccessListCatchUp::new(client.clone(), factory.clone(), Runtime::test());
632
633        // The list the peer served belongs to a block the reorg left canonical, so refusing it is
634        // the pivot check's doing.
635        assert!(matches!(
636            catch_up.next(write, PIVOT + 2).await,
637            Err(SnapSyncError::NonCanonicalBlock { block, hash })
638                if block == PIVOT + 2 && hash == chain.tip().hash
639        ));
640
641        assert_eq!(*client.block_requests(), [vec![chain.block(1).hash, chain.tip().hash]]);
642        let provider = factory.database_provider_ro().unwrap();
643        assert_eq!(provider.block_hash(PIVOT + 1).unwrap(), Some(chain.block(1).hash));
644        assert_eq!(provider.block_hash(PIVOT + 2).unwrap(), Some(replacement_hash));
645        assert_eq!(provider.catch_up_progress(write).unwrap().unwrap().applied(), chain.block(0));
646        assert_eq!(SnapStateSnapshot::read(&provider), before);
647    }
648
649    #[tokio::test]
650    async fn downloads_split_across_pivots_converge_on_the_latest_pivot_state() {
651        const FAR: B256 = B256::repeat_byte(0xfe);
652        let at = |balance| accounts_with(balance, [(FAR, account(3))]);
653        let chain = chain().with_state_roots(&[
654            (PIVOT + 1, state_root(&at(10))),
655            (PIVOT + 2, state_root(&at(20))),
656        ]);
657        assert_eq!(at(1).last().unwrap().0, FAR);
658        // The pivot serves every account but the last.
659        let (factory, write) = partially_downloaded(&chain, &at(1), 2);
660        let (_, mut catch_up) =
661            catch_up([chain.response(1, [Some(1)]), chain.response(2, [Some(2)])], factory.clone());
662
663        let write = advance(&factory, write, &chain, 1, &at(10));
664        applied(&mut catch_up, write, PIVOT + 3).await;
665        assert_eq!(balance(&factory), U256::from(10));
666        let write = advance(&factory, write, &chain, 2, &at(20));
667
668        // A range proved against the previous root cannot complete the new one's coverage.
669        let provider = factory.database_provider_rw().unwrap();
670        let origin = provider.account_coverage(write).unwrap().unwrap().next().unwrap();
671        let stale = verified_range(&at(10), 2..3, origin, &[origin, FAR]);
672        assert!(matches!(
673            provider.commit_account_range(write, &stale, Default::default(), Vec::new()),
674            Err(SnapSyncError::RootMismatch { .. })
675        ));
676        drop(provider);
677        download_rest(&factory, write, &at(20));
678
679        // Lists stop at the latest pivot, where the downloaded state now sits.
680        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
681        assert_eq!(progress.applied(), chain.block(2));
682        assert!(matches!(catch_up.next(write, PIVOT + 3).await.unwrap(), CatchUpStep::Complete));
683        assert_eq!(downloaded_root(&factory), state_root(&at(20)));
684    }
685
686    #[tokio::test]
687    async fn an_account_lists_reach_before_its_range_is_taken_from_the_new_root() {
688        let at = |balance| accounts_with(balance, []);
689        let chain = chain().with_state_roots(&[(PIVOT + 2, state_root(&at(20)))]);
690        // Only the account before the changed one is downloaded at the pivot.
691        assert_eq!(at(1)[1].0, keccak256(CHANGED));
692        let (factory, write) = partially_downloaded(&chain, &at(1), 1);
693        let write = advance(&factory, write, &chain, 2, &at(20));
694        let (_, mut catch_up) = catch_up([chain.response(1, [Some(1), Some(2)])], factory.clone());
695
696        // The lists have no base to change yet, so they leave the account to its range.
697        let progress = applied(&mut catch_up, write, PIVOT + 3).await;
698        assert_eq!(progress.applied(), chain.block(2));
699        let provider = factory.database_provider_ro().unwrap();
700        assert_eq!(
701            provider.tx_ref().get::<tables::HashedAccounts>(keccak256(CHANGED)).unwrap(),
702            None
703        );
704        drop(provider);
705
706        download_rest(&factory, write, &at(20));
707
708        assert_eq!(balance(&factory), U256::from(20));
709        assert_eq!(downloaded_root(&factory), state_root(&at(20)));
710    }
711
712    #[tokio::test]
713    async fn orphaned_lists_are_fetched_in_order_across_requests() {
714        let chain = chain();
715        let (orphaned, hashes) = orphaned(&chain);
716        let responses = [chain.response(1, [Some(1), Some(2)]), chain.response(2, [Some(3)])];
717        let (client, mut catch_up) = catch_up_with(responses, hashed_factory(), 2);
718
719        let lists = catch_up.orphaned_lists(&orphaned).await.unwrap().unwrap();
720
721        let lists: Vec<_> = lists.iter().map(|list| list.as_bal().to_vec()).collect();
722        assert_eq!(lists, [credit(10), credit(20), credit(30)]);
723        assert_eq!(*client.block_requests(), [hashes[..2].to_vec(), hashes[2..].to_vec()]);
724    }
725
726    #[tokio::test]
727    async fn a_missing_orphaned_list_fetches_nothing() {
728        let chain = chain();
729        let (orphaned, _) = orphaned(&chain);
730        // No peer holds the last orphaned block's list any more.
731        let responses = [chain.response(1, [Some(1), Some(2), None]), chain.response(2, [None])];
732        let (client, mut catch_up) = catch_up_with(responses, hashed_factory(), 3);
733
734        assert!(catch_up.orphaned_lists(&orphaned).await.unwrap().is_none());
735        assert_eq!(client.block_requests().len(), 2);
736    }
737
738    #[tokio::test]
739    async fn a_response_cut_short_keeps_its_lists_and_fetches_the_rest() {
740        let chain = chain();
741        let (orphaned, hashes) = orphaned(&chain);
742        // The first response stops at the byte limit after one list.
743        let responses = [chain.response(1, [Some(1)]), chain.response(2, [Some(2), Some(3)])];
744        let (client, mut catch_up) = catch_up_with(responses, hashed_factory(), 3);
745
746        let lists = catch_up.orphaned_lists(&orphaned).await.unwrap().unwrap();
747
748        let lists: Vec<_> = lists.iter().map(|list| list.as_bal().to_vec()).collect();
749        assert_eq!(lists, [credit(10), credit(20), credit(30)]);
750        assert_eq!(*client.block_requests(), [hashes.clone(), hashes[1..].to_vec()]);
751    }
752}
reth_snap_sync/catch_up/download.rs

reth_snap_sync/catch_up/
download.rs
