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

reth_snap_sync/catch_up/
download.rs
