reth_snap_sync/
reorg.rs

1//! Recovers downloaded state from a reorg that orphans the pivot.
2//!
3//! [EIP-8189](https://eips.ethereum.org/EIPS/eip-8189#synchronization-algorithm) repairs it from
4//! the abandoned branch's lists: every field they change is fetched again unless a list of the
5//! new branch overwrites it first. The reorg removes that branch's blocks from the canonical
6//! chain, so the attempt keeps what recovery needs of them from the moment it anchors.
7
8use crate::{common::SnapRecord, SnapSyncError};
9use alloy_eips::{eip1898::BlockWithParent, BlockNumHash};
10use alloy_primitives::{Sealable, B256};
11use reth_primitives_traits::{AlloyBlockHeader, SealedHeader};
12use reth_storage_api::{
13    BlockHashReader, HeaderProvider, MetadataProvider, MetadataWriter, SnapAttempt, SnapAttemptId,
14};
15use serde::{Deserialize, Serialize};
16
17// Blocks an attempt keeps through its pivot, bounding how deep a recoverable reorg can reach. A
18// deeper reorg restarts the attempt.
19const KEPT_BLOCKS: u64 = 64;
20
21/// Where a reorg left an attempt: the last block both branches share and the orphaned blocks
22/// after it.
23#[derive(Clone, Debug)]
24pub struct SnapReorg {
25    // Last block both branches share.
26    ancestor: BlockNumHash,
27    // Orphaned blocks through the pivot, oldest first. Empty while the pivot is canonical.
28    orphaned: Vec<BlockWithAccessListHash>,
29}
30
31impl SnapReorg {
32    /// Last block both branches share.
33    pub const fn ancestor(&self) -> BlockNumHash {
34        self.ancestor
35    }
36
37    /// Orphaned blocks through the pivot, oldest first.
38    pub fn orphaned(&self) -> &[BlockWithAccessListHash] {
39        &self.orphaned
40    }
41}
42
43/// A block an attempt keeps through its pivot, holding what recovery from a reorg reads of it.
44#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
45pub struct BlockWithAccessListHash {
46    // Block number and hash, with its parent's hash.
47    block: BlockWithParent,
48    // Block access list hash, absent before Amsterdam.
49    block_access_list_hash: Option<B256>,
50}
51
52impl BlockWithAccessListHash {
53    /// Block number and hash, with its parent's hash.
54    pub const fn block(&self) -> BlockWithParent {
55        self.block
56    }
57
58    /// Block access list hash, absent before Amsterdam.
59    pub const fn block_access_list_hash(&self) -> Option<B256> {
60        self.block_access_list_hash
61    }
62}
63
64impl<H: AlloyBlockHeader + Sealable> From<&SealedHeader<H>> for BlockWithAccessListHash {
65    fn from(header: &SealedHeader<H>) -> Self {
66        Self {
67            block: header.block_with_parent(),
68            block_access_list_hash: header.block_access_list_hash(),
69        }
70    }
71}
72
73// Blocks an attempt keeps through its pivot.
74#[derive(Serialize, Deserialize)]
75pub(crate) struct StoredAncestry {
76    // Encoding version, checked before the rest is decoded.
77    version: u32,
78    // Attempt the blocks belong to.
79    attempt: SnapAttemptId,
80    // Kept blocks, oldest first and ending at the pivot.
81    blocks: Vec<BlockWithAccessListHash>,
82}
83
84impl SnapRecord for StoredAncestry {
85    const KEY: &'static str = "snap_ancestry";
86    const VERSION: u32 = 2;
87}
88
89impl StoredAncestry {
90    // Keeps the canonical blocks of the last `KEPT_BLOCKS` through `pivot` for `attempt`. Keeps
91    // none unless they reach the pivot, leaving a reorg unrecoverable.
92    pub(crate) fn record<P: HeaderProvider + MetadataWriter>(
93        provider: &P,
94        attempt: SnapAttemptId,
95        pivot: BlockNumHash,
96    ) -> Result<(), SnapSyncError> {
97        let from = pivot.number.saturating_sub(KEPT_BLOCKS) + 1;
98        let headers = provider.sealed_headers_range(from..=pivot.number)?;
99        let contiguous = headers.windows(2).all(|pair| pair[1].parent_hash() == pair[0].hash());
100        if !contiguous || headers.last().map(SealedHeader::num_hash) != Some(pivot) {
101            return Self::clear(provider)
102        }
103        let blocks = headers.iter().map(BlockWithAccessListHash::from).collect();
104        Self { version: Self::VERSION, attempt, blocks }.write(provider)
105    }
106
107    // Finds where the canonical chain diverges from the blocks `attempt` keeps. `None` when they
108    // do not reach back to where the branches part.
109    pub(crate) fn reorg<P: MetadataProvider + BlockHashReader>(
110        provider: &P,
111        attempt: &SnapAttempt,
112    ) -> Result<Option<SnapReorg>, SnapSyncError> {
113        let Some(mut blocks) = Self::read(provider)?
114            .filter(|stored| stored.attempt == attempt.id())
115            .map(|stored| stored.blocks)
116        else {
117            return Ok(None)
118        };
119        // Kept blocks end at the pivot they were recorded for.
120        if blocks.last().map(|kept| kept.block.block) != Some(attempt.pivot()) {
121            return Ok(None)
122        }
123
124        // The highest kept block still canonical is where the branches part.
125        let mut split = None;
126        for (index, kept) in blocks.iter().enumerate().rev() {
127            let block = kept.block.block;
128            if provider.block_hash(block.number)? == Some(block.hash) {
129                split = Some((block, index + 1));
130                break
131            }
132        }
133        let (ancestor, from) = match split {
134            Some(split) => split,
135            None => {
136                // Otherwise they part just below the lowest kept block, or further down.
137                let lowest = blocks[0].block;
138                let Some(parent) = lowest.block.number.checked_sub(1) else { return Ok(None) };
139                if provider.block_hash(parent)? != Some(lowest.parent) {
140                    return Ok(None)
141                }
142                (BlockNumHash::new(parent, lowest.parent), 0)
143            }
144        };
145        Ok(Some(SnapReorg { ancestor, orphaned: blocks.split_off(from) }))
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use crate::{
153        test_utils::{account, hashed_factory, state_root, verified_range, BalChain},
154        SnapAccountStore, SnapAttemptStore, SnapCatchUpStore, SnapGeneration, SnapWrite,
155        StateRepairs,
156    };
157    use alloy_eip7928::{
158        AccountChanges, BalanceChange, BlockAccessIndex, NonceChange, SlotChanges, StorageChange,
159    };
160    use alloy_eips::eip7928::bal::{Bal, DecodedBal};
161    use alloy_primitives::{keccak256, Address, B256, U256};
162    use reth_provider::{
163        test_utils::{insert_headers, MockNodeTypesWithDB},
164        DatabaseProviderFactory, ProviderFactory,
165    };
166    use reth_storage_api::DBProvider;
167    use reth_trie_common::TrieAccount;
168
169    type Factory = ProviderFactory<MockNodeTypesWithDB>;
170
171    // Block both branches share.
172    const ANCESTOR: u64 = 2;
173
174    fn credit(address: Address, balance: u64) -> Vec<AccountChanges> {
175        vec![AccountChanges::new(address)
176            .with_balance_change(BalanceChange::new(BlockAccessIndex::new(1), U256::from(balance)))]
177    }
178
179    // Blocks 3 and 4 of the orphaned branch, and 3 through 5 of the one replacing it.
180    fn branches() -> (BalChain, BalChain) {
181        let address = Address::repeat_byte(0x11);
182        (
183            BalChain::new(ANCESTOR, [credit(address, 1), credit(address, 2)])
184                .with_state_root(root()),
185            BalChain::new(
186                ANCESTOR,
187                [credit(address, 10), credit(address, 20), credit(address, 30)],
188            )
189            .with_state_root(root()),
190        )
191    }
192
193    // Root of the accounts every chain commits to.
194    fn root() -> B256 {
195        state_root(&accounts().1)
196    }
197
198    // Three accounts, of which the first two are downloaded.
199    fn accounts() -> (Vec<Address>, Vec<(B256, TrieAccount)>) {
200        let mut addresses: Vec<_> = (1..=3).map(Address::repeat_byte).collect();
201        addresses.sort_by_key(|address| keccak256(address));
202        let accounts = addresses
203            .iter()
204            .enumerate()
205            .map(|(nonce, address)| (keccak256(address), account(nonce as u64)))
206            .collect();
207        (addresses, accounts)
208    }
209
210    // An attempt anchored at `chain`'s block `number`, with the first two accounts downloaded.
211    fn started(chain: &BalChain, number: u64) -> (Factory, SnapWrite) {
212        let (_, accounts) = accounts();
213        let factory = hashed_factory();
214        insert_headers(&factory, &chain.headers);
215        let provider = factory.database_provider_rw().unwrap();
216        let pivot = chain.headers[number as usize].num_hash();
217        let write =
218            provider.start_snap_attempt(SnapGeneration::new(pivot, state_root(&accounts))).unwrap();
219        provider.start_account_coverage(write).unwrap();
220        let range = verified_range(&accounts, 0..2, B256::ZERO, &[B256::ZERO, accounts[1].0]);
221        provider.commit_account_range(write, &range, Default::default(), Vec::new()).unwrap();
222        provider.commit().unwrap();
223        (factory, write)
224    }
225
226    fn reorg(factory: &Factory, write: SnapWrite) -> Option<SnapReorg> {
227        factory.database_provider_ro().unwrap().snap_reorg(write).unwrap()
228    }
229
230    fn kept(headers: &[SealedHeader]) -> Vec<BlockWithAccessListHash> {
231        headers.iter().map(BlockWithAccessListHash::from).collect()
232    }
233
234    // A verified list carrying `changes`.
235    fn list(changes: Vec<AccountChanges>) -> DecodedBal {
236        DecodedBal::from_rlp_bytes(alloy_rlp::encode(Bal::from(changes)).into()).unwrap()
237    }
238
239    #[test]
240    fn a_canonical_pivot_orphans_nothing() {
241        let (old, _) = branches();
242        let (factory, write) = started(&old, 4);
243
244        let reorg = reorg(&factory, write).unwrap();
245
246        assert_eq!(reorg.ancestor(), old.tip());
247        assert!(reorg.orphaned().is_empty());
248    }
249
250    #[test]
251    fn the_ancestor_is_the_last_block_both_branches_share() {
252        let (old, new) = branches();
253        let (factory, write) = started(&old, 4);
254        new.replace_after(&factory, ANCESTOR);
255
256        let reorg = reorg(&factory, write).unwrap();
257
258        assert_eq!(reorg.ancestor(), old.block(0));
259        assert_eq!(reorg.orphaned(), kept(&old.headers[3..]));
260    }
261
262    #[test]
263    fn the_ancestor_can_sit_just_below_the_kept_blocks() {
264        let address = Address::repeat_byte(0x11);
265        let old =
266            BalChain::new(0, [credit(address, 1), credit(address, 2)]).with_state_root(root());
267        let new = BalChain::new(0, [credit(address, 10)]).with_state_root(root());
268        let (factory, write) = started(&old, 2);
269        new.replace_after(&factory, 0);
270
271        let reorg = reorg(&factory, write).unwrap();
272
273        assert_eq!(reorg.ancestor(), old.block(0));
274        assert_eq!(reorg.orphaned(), kept(&old.headers[1..]));
275    }
276
277    #[test]
278    fn a_reorg_below_the_kept_blocks_is_unrecoverable() {
279        let address = Address::repeat_byte(0x11);
280        let depth = KEPT_BLOCKS + 2;
281        let old = BalChain::new(1, (0..depth).map(|n| credit(address, n))).with_state_root(root());
282        let new =
283            BalChain::new(1, (0..depth).map(|n| credit(address, n + 100))).with_state_root(root());
284        let (factory, write) = started(&old, 1 + depth);
285        new.replace_after(&factory, 1);
286
287        assert!(reorg(&factory, write).is_none());
288    }
289
290    #[test]
291    fn recovery_rewinds_catch_up_to_the_ancestor_and_refuses_old_writes() {
292        let (old, new) = branches();
293        let (factory, write) = started(&old, 4);
294        new.replace_after(&factory, ANCESTOR);
295        let (_, accounts) = accounts();
296        let generation = SnapGeneration::new(new.tip(), state_root(&accounts));
297
298        let provider = factory.database_provider_rw().unwrap();
299        let recovered =
300            provider.commit_reorg_recovery(write, old.block(0), &[], generation).unwrap();
301        provider.commit().unwrap();
302
303        let provider = factory.database_provider_ro().unwrap();
304        assert_eq!(provider.catch_up_progress(recovered).unwrap().unwrap().applied(), old.block(0));
305        assert!(matches!(
306            provider.authorize_snap_write(write),
307            Err(SnapSyncError::StaleWrite { .. })
308        ));
309        drop(provider);
310        // The kept blocks follow the new pivot.
311        let reorg = reorg(&factory, recovered).unwrap();
312        assert_eq!(reorg.ancestor(), new.tip());
313    }
314
315    #[test]
316    fn a_new_pivot_reorged_meanwhile_is_refused_without_scheduling() {
317        let (old, new) = branches();
318        let (factory, write) = started(&old, 4);
319        new.replace_after(&factory, ANCESTOR);
320        let (addresses, accounts) = accounts();
321        // The chain moved again, so the selected pivot is no longer canonical.
322        let generation = SnapGeneration::new(old.tip(), state_root(&accounts));
323        let lists = [list(credit(addresses[0], 5))];
324
325        let provider = factory.database_provider_rw().unwrap();
326        let refused = provider.commit_reorg_recovery(write, old.block(0), &lists, generation);
327        assert!(matches!(refused, Err(SnapSyncError::NonCanonicalBlock { block: 4, .. })));
328        drop(provider);
329
330        // The refused transaction is dropped, leaving the attempt to recover on the next pass.
331        let provider = factory.database_provider_ro().unwrap();
332        assert!(provider.snap_repairs(write).unwrap().is_empty());
333        assert_eq!(reorg(&factory, write).unwrap().ancestor(), old.block(0));
334    }
335
336    #[test]
337    fn catch_up_below_the_ancestor_stays_where_it_is() {
338        let (old, new) = branches();
339        let (factory, write) = started(&old, 1);
340        let (_, accounts) = accounts();
341        let provider = factory.database_provider_rw().unwrap();
342        let write = provider
343            .advance_snap_pivot(write, SnapGeneration::new(old.tip(), state_root(&accounts)))
344            .unwrap();
345        provider.commit().unwrap();
346        new.replace_after(&factory, ANCESTOR);
347
348        let provider = factory.database_provider_rw().unwrap();
349        let generation = SnapGeneration::new(new.tip(), state_root(&accounts));
350        let recovered =
351            provider.commit_reorg_recovery(write, old.block(0), &[], generation).unwrap();
352
353        assert_eq!(
354            provider.catch_up_progress(recovered).unwrap().unwrap().applied(),
355            old.headers[1].num_hash()
356        );
357    }
358
359    #[test]
360    fn orphaned_changes_to_downloaded_accounts_are_scheduled() {
361        let (old, new) = branches();
362        let (factory, write) = started(&old, 4);
363        new.replace_after(&factory, ANCESTOR);
364        let (addresses, accounts) = accounts();
365        let slot = SlotChanges::new(
366            U256::from(7),
367            vec![StorageChange::new(BlockAccessIndex::new(1), U256::ONE)],
368        );
369        let downloaded = credit(addresses[0], 5).remove(0).with_storage_change(slot);
370        let lists = [
371            list(vec![downloaded.clone()]),
372            list(vec![
373                // Only reads the account, which changes nothing.
374                AccountChanges::new(addresses[1]).with_storage_read(U256::ONE),
375                // Past the cursor, so downloaded whole later.
376                AccountChanges::new(addresses[2])
377                    .with_nonce_change(NonceChange::new(BlockAccessIndex::new(1), 1)),
378            ]),
379        ];
380        let generation = SnapGeneration::new(new.tip(), state_root(&accounts));
381
382        let provider = factory.database_provider_rw().unwrap();
383        let recovered =
384            provider.commit_reorg_recovery(write, old.block(0), &lists, generation).unwrap();
385
386        let mut expected = StateRepairs::default();
387        expected.insert_changes(keccak256(addresses[0]), &downloaded);
388        assert_eq!(provider.snap_repairs(recovered).unwrap(), expected);
389    }
390}
reth_snap_sync/reorg.rs

reth_snap_sync/
reorg.rs
