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

reth_snap_sync/
reorg.rs
