Skip to main content

reth_prune/segments/user/
account_history.rs

1use crate::{
2    db_ext::DbTxPruneExt,
3    segments::{
4        user::history::{finalize_history_prune, HistoryPruneResult},
5        PruneInput, Segment,
6    },
7    PrunerError,
8};
9use alloy_primitives::{map::HashMap, BlockNumber};
10use reth_db_api::{models::ShardedKey, tables, transaction::DbTxMut};
11use reth_provider::{
12    changeset_walker::StaticFileAccountChangesetWalker, DBProvider, EitherWriter,
13    RocksDBProviderFactory, StaticFileProviderFactory,
14};
15use reth_prune_types::{
16    PruneMode, PrunePurpose, PruneSegment, SegmentOutput, SegmentOutputCheckpoint,
17};
18use reth_static_file_types::StaticFileSegment;
19use reth_storage_api::{ChangeSetReader, StorageSettingsCache};
20use tracing::{instrument, trace};
21
22/// Number of account history tables to prune in one step.
23///
24/// Account History consists of two tables: [`tables::AccountChangeSets`] (either in database or
25/// static files) and [`tables::AccountsHistory`]. We want to prune them to the same block number.
26const ACCOUNT_HISTORY_TABLES_TO_PRUNE: usize = 2;
27
28#[derive(Debug)]
29pub struct AccountHistory {
30    mode: PruneMode,
31}
32
33impl AccountHistory {
34    pub const fn new(mode: PruneMode) -> Self {
35        Self { mode }
36    }
37}
38
39impl<Provider> Segment<Provider> for AccountHistory
40where
41    Provider: DBProvider<Tx: DbTxMut>
42        + StaticFileProviderFactory
43        + StorageSettingsCache
44        + ChangeSetReader
45        + RocksDBProviderFactory,
46{
47    fn segment(&self) -> PruneSegment {
48        PruneSegment::AccountHistory
49    }
50
51    fn mode(&self) -> Option<PruneMode> {
52        Some(self.mode)
53    }
54
55    fn purpose(&self) -> PrunePurpose {
56        PrunePurpose::User
57    }
58
59    #[instrument(
60        name = "AccountHistory::prune",
61        target = "pruner",
62        skip(self, provider),
63        ret(level = "trace")
64    )]
65    fn prune(&self, provider: &Provider, input: PruneInput) -> Result<SegmentOutput, PrunerError> {
66        let range = match input.get_next_block_range() {
67            Some(range) => range,
68            None => {
69                trace!(target: "pruner", "No account history to prune");
70                return Ok(SegmentOutput::done())
71            }
72        };
73        let range_end = *range.end();
74
75        // Check where account history indices are stored
76        if provider.cached_storage_settings().storage_v2 {
77            return self.prune_rocksdb(provider, input, range, range_end);
78        }
79
80        // Check where account changesets are stored (MDBX path)
81        if EitherWriter::account_changesets_destination(provider).is_static_file() {
82            self.prune_static_files(provider, input, range, range_end)
83        } else {
84            self.prune_database(provider, input, range, range_end)
85        }
86    }
87}
88
89impl AccountHistory {
90    /// Prunes account history when changesets are stored in static files.
91    fn prune_static_files<Provider>(
92        &self,
93        provider: &Provider,
94        input: PruneInput,
95        range: std::ops::RangeInclusive<BlockNumber>,
96        range_end: BlockNumber,
97    ) -> Result<SegmentOutput, PrunerError>
98    where
99        Provider: DBProvider<Tx: DbTxMut> + StaticFileProviderFactory + ChangeSetReader,
100    {
101        let mut limiter = if let Some(limit) = input.limiter.deleted_entries_limit() {
102            input.limiter.set_deleted_entries_limit(limit / ACCOUNT_HISTORY_TABLES_TO_PRUNE)
103        } else {
104            input.limiter
105        };
106
107        // The limiter may already be exhausted from a previous segment in the same prune run.
108        // Early exit avoids unnecessary iteration when no budget remains.
109        if limiter.is_limit_reached() {
110            return Ok(SegmentOutput::not_done(
111                limiter.interrupt_reason(),
112                input.previous_checkpoint.map(SegmentOutputCheckpoint::from_prune_checkpoint),
113            ))
114        }
115
116        // Deleted account changeset keys (account addresses) with the highest block number deleted
117        // for that key.
118        //
119        // The size of this map is limited by `prune_delete_limit * blocks_since_last_run /
120        // ACCOUNT_HISTORY_TABLES_TO_PRUNE`, and with current default it's usually `3500 * 5
121        // / 2`, so 8750 entries. Each entry is `160 bit + 64 bit`, so the total
122        // size should be up to ~0.25MB + some hashmap overhead. `blocks_since_last_run` is
123        // additionally limited by the `max_reorg_depth`, so no OOM is expected here.
124        let mut highest_deleted_accounts = HashMap::default();
125        let mut last_changeset_pruned_block = None;
126        let mut pruned_changesets = 0;
127        let mut done = true;
128
129        let walker = StaticFileAccountChangesetWalker::new(provider, range);
130        for result in walker {
131            let (block_number, changeset) = result?;
132            // The walk itself deletes nothing, so an interrupted block cannot be resumed: giving
133            // up the budget inside block N reports checkpoint N-1 and the next run rereads the
134            // same entries, forever. Stop on block boundaries only, overshooting the budget by at
135            // most the rest of one block.
136            if limiter.is_limit_reached() &&
137                last_changeset_pruned_block.is_some_and(|last| last != block_number)
138            {
139                done = false;
140                break;
141            }
142            highest_deleted_accounts.insert(changeset.address, block_number);
143            last_changeset_pruned_block = Some(block_number);
144            pruned_changesets += 1;
145            limiter.increment_deleted_entries_count();
146        }
147
148        // Delete static file jars only when fully processed
149        if done && let Some(last_block) = last_changeset_pruned_block {
150            provider
151                .static_file_provider()
152                .delete_segment_below_block(StaticFileSegment::AccountChangeSets, last_block + 1)?;
153        }
154        trace!(target: "pruner", pruned = %pruned_changesets, %done, "Pruned account history (changesets from static files)");
155
156        let result = HistoryPruneResult {
157            highest_deleted: highest_deleted_accounts,
158            last_pruned_block: last_changeset_pruned_block,
159            pruned_count: pruned_changesets,
160            done,
161        };
162        finalize_history_prune::<_, tables::AccountsHistory, _, _>(
163            provider,
164            result,
165            range_end,
166            &limiter,
167            ShardedKey::new,
168            |a, b| a.key == b.key,
169        )
170        .map_err(Into::into)
171    }
172
173    fn prune_database<Provider>(
174        &self,
175        provider: &Provider,
176        input: PruneInput,
177        range: std::ops::RangeInclusive<BlockNumber>,
178        range_end: BlockNumber,
179    ) -> Result<SegmentOutput, PrunerError>
180    where
181        Provider: DBProvider<Tx: DbTxMut>,
182    {
183        let mut limiter = if let Some(limit) = input.limiter.deleted_entries_limit() {
184            input.limiter.set_deleted_entries_limit(limit / ACCOUNT_HISTORY_TABLES_TO_PRUNE)
185        } else {
186            input.limiter
187        };
188
189        if limiter.is_limit_reached() {
190            return Ok(SegmentOutput::not_done(
191                limiter.interrupt_reason(),
192                input.previous_checkpoint.map(SegmentOutputCheckpoint::from_prune_checkpoint),
193            ))
194        }
195
196        // Deleted account changeset keys (account addresses) with the highest block number deleted
197        // for that key.
198        //
199        // The size of this map is limited by `prune_delete_limit * blocks_since_last_run /
200        // ACCOUNT_HISTORY_TABLES_TO_PRUNE`, and with the current defaults it's usually `3500 * 5 /
201        // 2`, so 8750 entries. Each entry is `160 bit + 64 bit`, so the total size should be up to
202        // ~0.25MB + some hashmap overhead. `blocks_since_last_run` is additionally limited by the
203        // `max_reorg_depth`, so no OOM is expected here.
204        let mut last_changeset_pruned_block = None;
205        let mut highest_deleted_accounts = HashMap::default();
206        let (pruned_changesets, done) =
207            provider.tx_ref().prune_table_with_range::<tables::AccountChangeSets>(
208                range,
209                &mut limiter,
210                |_| false,
211                |(block_number, account)| {
212                    highest_deleted_accounts.insert(account.address, block_number);
213                    last_changeset_pruned_block = Some(block_number);
214                },
215            )?;
216        trace!(target: "pruner", pruned = %pruned_changesets, %done, "Pruned account history (changesets from database)");
217
218        // The table walk can stop in the middle of a block, so the interrupted block has to be
219        // pruned again on the next run.
220        let last_pruned_block = last_changeset_pruned_block.map(|block_number| {
221            if done {
222                block_number
223            } else {
224                block_number.saturating_sub(1)
225            }
226        });
227
228        let result = HistoryPruneResult {
229            highest_deleted: highest_deleted_accounts,
230            last_pruned_block,
231            pruned_count: pruned_changesets,
232            done,
233        };
234        finalize_history_prune::<_, tables::AccountsHistory, _, _>(
235            provider,
236            result,
237            range_end,
238            &limiter,
239            ShardedKey::new,
240            |a, b| a.key == b.key,
241        )
242        .map_err(Into::into)
243    }
244
245    /// Prunes account history when indices are stored in `RocksDB`.
246    ///
247    /// Reads account changesets from static files and prunes the corresponding
248    /// `RocksDB` history shards.
249    fn prune_rocksdb<Provider>(
250        &self,
251        provider: &Provider,
252        input: PruneInput,
253        range: std::ops::RangeInclusive<BlockNumber>,
254        range_end: BlockNumber,
255    ) -> Result<SegmentOutput, PrunerError>
256    where
257        Provider: DBProvider + StaticFileProviderFactory + ChangeSetReader + RocksDBProviderFactory,
258    {
259        // Unlike MDBX path, we don't divide the limit by 2 because RocksDB path only prunes
260        // history shards (no separate changeset table to delete from). The changesets are in
261        // static files which are deleted separately.
262        let mut limiter = input.limiter;
263
264        if limiter.is_limit_reached() {
265            return Ok(SegmentOutput::not_done(
266                limiter.interrupt_reason(),
267                input.previous_checkpoint.map(SegmentOutputCheckpoint::from_prune_checkpoint),
268            ))
269        }
270
271        let mut highest_deleted_accounts = HashMap::<_, _>::default();
272        let mut last_changeset_pruned_block = None;
273        let mut changesets_processed = 0usize;
274        let mut done = true;
275
276        // Walk account changesets from static files using a streaming iterator.
277        // For each changeset, track the highest block number seen for each address
278        // to determine which history shard entries need pruning.
279        let walker = StaticFileAccountChangesetWalker::new(provider, range);
280        for result in walker {
281            let (block_number, changeset) = result?;
282            // Static file changesets are not deleted here, so an interrupted block cannot be
283            // resumed: giving up the budget inside block N reports checkpoint N-1 and the next run
284            // rereads the same entries, forever. Stop on block boundaries only, overshooting the
285            // budget by at most the rest of one block.
286            if limiter.is_limit_reached() &&
287                last_changeset_pruned_block.is_some_and(|last| last != block_number)
288            {
289                done = false;
290                break;
291            }
292            highest_deleted_accounts.insert(changeset.address, block_number);
293            last_changeset_pruned_block = Some(block_number);
294            changesets_processed += 1;
295            limiter.increment_deleted_entries_count();
296        }
297        trace!(target: "pruner", processed = %changesets_processed, %done, "Scanned account changesets from static files");
298
299        let last_changeset_pruned_block = last_changeset_pruned_block.unwrap_or(range_end);
300
301        // Prune RocksDB history shards for affected accounts
302        let mut deleted_shards = 0usize;
303        let mut updated_shards = 0usize;
304
305        // Sort by address for better RocksDB cache locality
306        let mut sorted_accounts: Vec<_> = highest_deleted_accounts.into_iter().collect();
307        sorted_accounts.sort_unstable_by_key(|(addr, _)| *addr);
308
309        provider.with_rocksdb_batch(|mut batch| {
310            let targets: Vec<_> = sorted_accounts
311                .iter()
312                .map(|(addr, highest)| (*addr, (*highest).min(last_changeset_pruned_block)))
313                .collect();
314
315            let outcomes = batch.prune_account_history_batch(&targets)?;
316            deleted_shards = outcomes.deleted;
317            updated_shards = outcomes.updated;
318
319            Ok(((), Some(batch.into_inner())))
320        })?;
321        trace!(target: "pruner", deleted = deleted_shards, updated = updated_shards, %done, "Pruned account history (RocksDB indices)");
322
323        // Delete static file jars only when fully processed. During provider.commit(), RocksDB
324        // batch is committed before the MDBX checkpoint. If crash occurs after RocksDB commit
325        // but before MDBX commit, on restart the pruner checkpoint indicates data needs
326        // re-pruning, but the RocksDB shards are already pruned - this is safe because pruning
327        // is idempotent (re-pruning already-pruned shards is a no-op).
328        if done {
329            provider.static_file_provider().delete_segment_below_block(
330                StaticFileSegment::AccountChangeSets,
331                last_changeset_pruned_block + 1,
332            )?;
333        }
334
335        let progress = limiter.progress(done);
336
337        Ok(SegmentOutput {
338            progress,
339            pruned: changesets_processed + deleted_shards + updated_shards,
340            checkpoint: Some(SegmentOutputCheckpoint {
341                block_number: Some(last_changeset_pruned_block),
342                tx_number: None,
343            }),
344        })
345    }
346}
347
348#[cfg(test)]
349mod tests {
350    use super::ACCOUNT_HISTORY_TABLES_TO_PRUNE;
351    use crate::segments::{AccountHistory, PruneInput, PruneLimiter, Segment, SegmentOutput};
352    use alloy_primitives::{Address, BlockNumber, B256, U256};
353    use assert_matches::assert_matches;
354    use reth_db_api::{
355        models::{ShardedKey, StorageSettings},
356        tables, BlockNumberList,
357    };
358    use reth_primitives_traits::Account;
359    use reth_provider::{
360        DBProvider, DatabaseProviderFactory, PruneCheckpointReader, RocksDBProviderFactory,
361        StaticFileProviderFactory,
362    };
363    use reth_prune_types::{
364        PruneCheckpoint, PruneInterruptReason, PruneMode, PruneProgress, PruneSegment,
365    };
366    use reth_stages::test_utils::{StorageKind, TestStageDB};
367    use reth_storage_api::StorageSettingsCache;
368    use reth_testing_utils::generators::{
369        self, random_block_range, random_changeset_range, random_eoa_accounts, BlockRangeParams,
370        ChangeSet,
371    };
372    use std::{collections::BTreeMap, ops::AddAssign};
373
374    #[test]
375    fn prune_legacy() {
376        let db = TestStageDB::default();
377        let mut rng = generators::rng();
378
379        let blocks = random_block_range(
380            &mut rng,
381            0..=5000,
382            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
383        );
384        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
385
386        let accounts = random_eoa_accounts(&mut rng, 2).into_iter().collect::<BTreeMap<_, _>>();
387
388        let (changesets, _) = random_changeset_range(
389            &mut rng,
390            blocks.iter(),
391            accounts.into_iter().map(|(addr, acc)| (addr, (acc, Vec::new()))),
392            0..0,
393            0..0,
394        );
395        db.insert_changesets(changesets.clone(), None).expect("insert changesets");
396        db.insert_history(changesets.clone(), None).expect("insert history");
397
398        let account_occurrences = db.table::<tables::AccountsHistory>().unwrap().into_iter().fold(
399            BTreeMap::<_, usize>::new(),
400            |mut map, (key, _)| {
401                map.entry(key.key).or_default().add_assign(1);
402                map
403            },
404        );
405        assert!(account_occurrences.into_iter().any(|(_, occurrences)| occurrences > 1));
406
407        assert_eq!(
408            db.table::<tables::AccountChangeSets>().unwrap().len(),
409            changesets.iter().flatten().count()
410        );
411
412        let original_shards = db.table::<tables::AccountsHistory>().unwrap();
413
414        let test_prune =
415            |to_block: BlockNumber, run: usize, expected_result: (PruneProgress, usize)| {
416                let prune_mode = PruneMode::Before(to_block);
417                let deleted_entries_limit = 2000;
418                let mut limiter =
419                    PruneLimiter::default().set_deleted_entries_limit(deleted_entries_limit);
420                let input = PruneInput {
421                    previous_checkpoint: db
422                        .factory
423                        .provider()
424                        .unwrap()
425                        .get_prune_checkpoint(PruneSegment::AccountHistory)
426                        .unwrap(),
427                    to_block,
428                    limiter: limiter.clone(),
429                };
430                let segment = AccountHistory::new(prune_mode);
431
432                let provider = db.factory.database_provider_rw().unwrap();
433                provider.set_storage_settings_cache(StorageSettings::v1());
434                let result = segment.prune(&provider, input).unwrap();
435                limiter.increment_deleted_entries_count_by(result.pruned);
436
437                assert_matches!(
438                    result,
439                    SegmentOutput {progress, pruned, checkpoint: Some(_)}
440                        if (progress, pruned) == expected_result
441                );
442
443                segment
444                    .save_checkpoint(
445                        &provider,
446                        result.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
447                    )
448                    .unwrap();
449                provider.commit().expect("commit");
450
451                let changesets = changesets
452                    .iter()
453                    .enumerate()
454                    .flat_map(|(block_number, changeset)| {
455                        changeset.iter().map(move |change| (block_number, change))
456                    })
457                    .collect::<Vec<_>>();
458
459                #[expect(clippy::skip_while_next)]
460                let pruned = changesets
461                    .iter()
462                    .enumerate()
463                    .skip_while(|(i, (block_number, _))| {
464                        *i < deleted_entries_limit / ACCOUNT_HISTORY_TABLES_TO_PRUNE * run &&
465                            *block_number <= to_block as usize
466                    })
467                    .next()
468                    .map(|(i, _)| i)
469                    .unwrap_or_default();
470
471                // Skip what we've pruned so far, subtracting one to get last pruned block number
472                // further down
473                let mut pruned_changesets = changesets.iter().skip(pruned.saturating_sub(1));
474
475                let last_pruned_block_number = pruned_changesets
476                    .next()
477                    .map(|(block_number, _)| if result.progress.is_finished() {
478                        *block_number
479                    } else {
480                        block_number.saturating_sub(1)
481                    } as BlockNumber)
482                    .unwrap_or(to_block);
483
484                let pruned_changesets = pruned_changesets.fold(
485                    BTreeMap::<_, Vec<_>>::new(),
486                    |mut acc, (block_number, change)| {
487                        acc.entry(block_number).or_default().push(change);
488                        acc
489                    },
490                );
491
492                assert_eq!(
493                    db.table::<tables::AccountChangeSets>().unwrap().len(),
494                    pruned_changesets.values().flatten().count()
495                );
496
497                let actual_shards = db.table::<tables::AccountsHistory>().unwrap();
498
499                let expected_shards = original_shards
500                    .iter()
501                    .filter(|(key, _)| key.highest_block_number > last_pruned_block_number)
502                    .map(|(key, blocks)| {
503                        let new_blocks =
504                            blocks.iter().skip_while(|block| *block <= last_pruned_block_number);
505                        (key.clone(), BlockNumberList::new_pre_sorted(new_blocks))
506                    })
507                    .collect::<Vec<_>>();
508
509                assert_eq!(actual_shards, expected_shards);
510
511                assert_eq!(
512                    db.factory
513                        .provider()
514                        .unwrap()
515                        .get_prune_checkpoint(PruneSegment::AccountHistory)
516                        .unwrap(),
517                    Some(PruneCheckpoint {
518                        block_number: Some(last_pruned_block_number),
519                        tx_number: None,
520                        prune_mode
521                    })
522                );
523            };
524
525        test_prune(
526            998,
527            1,
528            (PruneProgress::HasMoreData(PruneInterruptReason::DeletedEntriesLimitReached), 1000),
529        );
530        test_prune(998, 2, (PruneProgress::Finished, 998));
531        test_prune(1400, 3, (PruneProgress::Finished, 804));
532    }
533
534    #[test]
535    fn prune_rocksdb_path() {
536        let db = TestStageDB::default();
537        let mut rng = generators::rng();
538
539        let blocks = random_block_range(
540            &mut rng,
541            0..=100,
542            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
543        );
544        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
545
546        let accounts = random_eoa_accounts(&mut rng, 2).into_iter().collect::<BTreeMap<_, _>>();
547
548        let (changesets, _) = random_changeset_range(
549            &mut rng,
550            blocks.iter(),
551            accounts.into_iter().map(|(addr, acc)| (addr, (acc, Vec::new()))),
552            0..0,
553            0..0,
554        );
555
556        db.insert_changesets_to_static_files(changesets.clone(), None)
557            .expect("insert changesets to static files");
558
559        let mut account_blocks: BTreeMap<_, Vec<u64>> = BTreeMap::new();
560        for (block, changeset) in changesets.iter().enumerate() {
561            for (address, _, _) in changeset {
562                account_blocks.entry(*address).or_default().push(block as u64);
563            }
564        }
565
566        let rocksdb = db.factory.rocksdb_provider();
567        let mut batch = rocksdb.batch();
568        for (address, block_numbers) in &account_blocks {
569            let shard = BlockNumberList::new_pre_sorted(block_numbers.iter().copied());
570            batch
571                .put::<tables::AccountsHistory>(ShardedKey::new(*address, u64::MAX), &shard)
572                .unwrap();
573        }
574        batch.commit().unwrap();
575
576        for (address, expected_blocks) in &account_blocks {
577            let shards = rocksdb.account_history_shards(*address).unwrap();
578            assert_eq!(shards.len(), 1);
579            assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), *expected_blocks);
580        }
581
582        let to_block: BlockNumber = 50;
583        let prune_mode = PruneMode::Before(to_block);
584        let input =
585            PruneInput { previous_checkpoint: None, to_block, limiter: PruneLimiter::default() };
586        let segment = AccountHistory::new(prune_mode);
587
588        db.factory.set_storage_settings_cache(StorageSettings::v2());
589
590        let provider = db.factory.database_provider_rw().unwrap();
591        let result = segment.prune(&provider, input).unwrap();
592        provider.commit().expect("commit");
593
594        assert_matches!(
595            result,
596            SegmentOutput { progress: PruneProgress::Finished, pruned, checkpoint: Some(_) }
597                if pruned > 0
598        );
599
600        for (address, original_blocks) in &account_blocks {
601            let shards = rocksdb.account_history_shards(*address).unwrap();
602
603            let expected_blocks: Vec<u64> =
604                original_blocks.iter().copied().filter(|b| *b > to_block).collect();
605
606            if expected_blocks.is_empty() {
607                assert!(
608                    shards.is_empty(),
609                    "Expected no shards for address {address:?} after pruning"
610                );
611            } else {
612                assert_eq!(shards.len(), 1, "Expected 1 shard for address {address:?}");
613                assert_eq!(
614                    shards[0].1.iter().collect::<Vec<_>>(),
615                    expected_blocks,
616                    "Shard blocks mismatch for address {address:?}"
617                );
618            }
619        }
620
621        let static_file_provider = db.factory.static_file_provider();
622        let highest_block = static_file_provider.get_highest_static_file_block(
623            reth_static_file_types::StaticFileSegment::AccountChangeSets,
624        );
625        if let Some(block) = highest_block {
626            assert!(
627                block > to_block,
628                "Static files should only contain blocks above to_block ({to_block}), got {block}"
629            );
630        }
631    }
632
633    /// Tests that when a limiter stops mid-block (with multiple changes for the same block),
634    /// the checkpoint is set to `block_number - 1` to avoid dangling index entries.
635    #[test]
636    #[allow(clippy::clone_on_copy)]
637    fn prune_partial_progress_mid_block() {
638        let db = TestStageDB::default();
639        let mut rng = generators::rng();
640
641        // Create blocks 0..=10
642        let blocks = random_block_range(
643            &mut rng,
644            0..=10,
645            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
646        );
647        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
648
649        // Create specific changesets where block 5 has 4 account changes
650        let addr1 = Address::with_last_byte(1);
651        let addr2 = Address::with_last_byte(2);
652        let addr3 = Address::with_last_byte(3);
653        let addr4 = Address::with_last_byte(4);
654        let addr5 = Address::with_last_byte(5);
655
656        let account = Account { nonce: 1, balance: U256::from(100), ..Default::default() };
657
658        // Build changesets: blocks 0-4 have 1 change each, block 5 has 4 changes, block 6 has 1
659        let changesets: Vec<ChangeSet> = vec![
660            vec![(addr1, account.clone(), vec![])], // block 0
661            vec![(addr1, account.clone(), vec![])], // block 1
662            vec![(addr1, account.clone(), vec![])], // block 2
663            vec![(addr1, account.clone(), vec![])], // block 3
664            vec![(addr1, account.clone(), vec![])], // block 4
665            // block 5: 4 different account changes (sorted by address for consistency)
666            vec![
667                (addr1, account.clone(), vec![]),
668                (addr2, account.clone(), vec![]),
669                (addr3, account.clone(), vec![]),
670                (addr4, account.clone(), vec![]),
671            ],
672            vec![(addr5, account, vec![])], // block 6
673        ];
674
675        db.insert_changesets(changesets.clone(), None).expect("insert changesets");
676        db.insert_history(changesets.clone(), None).expect("insert history");
677
678        // Total changesets: 5 (blocks 0-4) + 4 (block 5) + 1 (block 6) = 10
679        assert_eq!(
680            db.table::<tables::AccountChangeSets>().unwrap().len(),
681            changesets.iter().flatten().count()
682        );
683
684        let prune_mode = PruneMode::Before(10);
685
686        // Set limiter to stop after 7 entries (mid-block 5: 5 from blocks 0-4, then 2 of 4 from
687        // block 5). Due to ACCOUNT_HISTORY_TABLES_TO_PRUNE=2, actual limit is 7/2=3
688        // changesets. So we'll process blocks 0, 1, 2 (3 changesets), stopping before block
689        // 3. Actually, let's use a higher limit to reach block 5. With limit=14, we get 7
690        // changeset slots. Blocks 0-4 use 5 slots, leaving 2 for block 5 (which has 4), so
691        // we stop mid-block 5.
692        let deleted_entries_limit = 14; // 14/2 = 7 changeset entries before limit
693        let limiter = PruneLimiter::default().set_deleted_entries_limit(deleted_entries_limit);
694
695        let input = PruneInput { previous_checkpoint: None, to_block: 10, limiter };
696        let segment = AccountHistory::new(prune_mode);
697
698        let provider = db.factory.database_provider_rw().unwrap();
699        provider.set_storage_settings_cache(StorageSettings::v1());
700        let result = segment.prune(&provider, input).unwrap();
701
702        // Should report that there's more data
703        assert!(!result.progress.is_finished(), "Expected HasMoreData since we stopped mid-block");
704
705        // Save checkpoint and commit
706        segment
707            .save_checkpoint(&provider, result.checkpoint.unwrap().as_prune_checkpoint(prune_mode))
708            .unwrap();
709        provider.commit().expect("commit");
710
711        // Verify checkpoint is set to block 4 (not 5), since block 5 is incomplete
712        let checkpoint = db
713            .factory
714            .provider()
715            .unwrap()
716            .get_prune_checkpoint(PruneSegment::AccountHistory)
717            .unwrap()
718            .expect("checkpoint should exist");
719
720        assert_eq!(
721            checkpoint.block_number,
722            Some(4),
723            "Checkpoint should be block 4 (block before incomplete block 5)"
724        );
725
726        // Verify remaining changesets (block 5 and 6 should still have entries)
727        let remaining_changesets = db.table::<tables::AccountChangeSets>().unwrap();
728        // After pruning blocks 0-4, remaining should be block 5 (4 entries) + block 6 (1 entry) = 5
729        // But since we stopped mid-block 5, some of block 5 might be pruned
730        // However, checkpoint is 4, so on re-run we should re-process from block 5
731        assert!(
732            !remaining_changesets.is_empty(),
733            "Should have remaining changesets for blocks 5-6"
734        );
735
736        // Verify no dangling history indices for blocks that weren't fully pruned
737        // The indices for block 5 should still reference blocks <= 5 appropriately
738        let history = db.table::<tables::AccountsHistory>().unwrap();
739        for (key, _blocks) in &history {
740            // All blocks in the history should be > checkpoint block number
741            // OR the shard's highest_block_number should be > checkpoint
742            assert!(
743                key.highest_block_number > 4,
744                "Found stale history shard with highest_block_number {} <= checkpoint 4",
745                key.highest_block_number
746            );
747        }
748
749        // Run prune again to complete - should finish processing block 5 and 6
750        let input2 = PruneInput {
751            previous_checkpoint: Some(checkpoint),
752            to_block: 10,
753            limiter: PruneLimiter::default().set_deleted_entries_limit(100), // high limit
754        };
755
756        let provider2 = db.factory.database_provider_rw().unwrap();
757        provider2.set_storage_settings_cache(StorageSettings::v1());
758        let result2 = segment.prune(&provider2, input2).unwrap();
759
760        assert!(result2.progress.is_finished(), "Second run should complete");
761
762        segment
763            .save_checkpoint(
764                &provider2,
765                result2.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
766            )
767            .unwrap();
768        provider2.commit().expect("commit");
769
770        // Verify final checkpoint
771        let final_checkpoint = db
772            .factory
773            .provider()
774            .unwrap()
775            .get_prune_checkpoint(PruneSegment::AccountHistory)
776            .unwrap()
777            .expect("checkpoint should exist");
778
779        // Should now be at block 6 (the last block with changesets)
780        assert_eq!(final_checkpoint.block_number, Some(6), "Final checkpoint should be at block 6");
781
782        // All changesets should be pruned
783        let final_changesets = db.table::<tables::AccountChangeSets>().unwrap();
784        assert!(final_changesets.is_empty(), "All changesets up to block 10 should be pruned");
785    }
786
787    /// A block holding at least a whole run's budget of changesets must not stall the `RocksDB`
788    /// path: the walk deletes no changesets, so a checkpoint rewound below such a block would make
789    /// every later run reread it and never advance.
790    #[test]
791    fn dense_block_advances_rocksdb_checkpoint() {
792        let db = TestStageDB::default();
793        let mut rng = generators::rng();
794
795        let blocks = random_block_range(
796            &mut rng,
797            0..=20,
798            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
799        );
800        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
801
802        let accounts = random_eoa_accounts(&mut rng, 2).into_iter().collect::<BTreeMap<_, _>>();
803        let (changesets, _) = random_changeset_range(
804            &mut rng,
805            blocks.iter(),
806            accounts.into_iter().map(|(addr, acc)| (addr, (acc, Vec::new()))),
807            0..0,
808            0..0,
809        );
810        // `random_changeset_range` emits exactly 2 account changesets per block (sender +
811        // recipient), so a budget of 2 makes every block "dense".
812        assert!(changesets.iter().all(|changeset| changeset.len() == 2));
813
814        db.insert_changesets_to_static_files(changesets.clone(), None)
815            .expect("insert changesets to static files");
816
817        // History index lives in RocksDB on the v2 path.
818        let mut account_blocks: BTreeMap<_, Vec<u64>> = BTreeMap::new();
819        for (block, changeset) in changesets.iter().enumerate() {
820            for (address, _, _) in changeset {
821                account_blocks.entry(*address).or_default().push(block as u64);
822            }
823        }
824        let rocksdb = db.factory.rocksdb_provider();
825        let mut batch = rocksdb.batch();
826        for (address, block_numbers) in &account_blocks {
827            let shard = BlockNumberList::new_pre_sorted(block_numbers.iter().copied());
828            batch
829                .put::<tables::AccountsHistory>(ShardedKey::new(*address, u64::MAX), &shard)
830                .unwrap();
831        }
832        batch.commit().unwrap();
833
834        db.factory.set_storage_settings_cache(StorageSettings::v2());
835
836        let to_block: BlockNumber = 15;
837        let prune_mode = PruneMode::Before(to_block);
838        let segment = AccountHistory::new(prune_mode);
839
840        // Start from a checkpoint in the middle so a rewind can't be masked by block 0.
841        let mut checkpoint = PruneCheckpoint { block_number: Some(4), tx_number: None, prune_mode };
842
843        let run_prune = |checkpoint: PruneCheckpoint, limit: usize| {
844            let input = PruneInput {
845                previous_checkpoint: Some(checkpoint),
846                to_block,
847                limiter: PruneLimiter::default().set_deleted_entries_limit(limit),
848            };
849
850            let provider = db.factory.database_provider_rw().unwrap();
851            provider.set_storage_settings_cache(StorageSettings::v2());
852            let result = segment.prune(&provider, input).unwrap();
853            segment
854                .save_checkpoint(
855                    &provider,
856                    result.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
857                )
858                .unwrap();
859            provider.commit().expect("commit");
860
861            let checkpoint = db
862                .factory
863                .provider()
864                .unwrap()
865                .get_prune_checkpoint(PruneSegment::AccountHistory)
866                .unwrap()
867                .unwrap();
868            (result, checkpoint)
869        };
870
871        // The RocksDB path does not halve the limit, so a budget of 2 is exactly one dense block.
872        for _ in 0..3 {
873            let previous = checkpoint.block_number;
874            let (result, next) = run_prune(checkpoint, 2);
875            checkpoint = next;
876
877            assert!(
878                !result.progress.is_finished(),
879                "the range is longer than one run's budget allows"
880            );
881            assert!(
882                checkpoint.block_number > previous,
883                "checkpoint must advance past the dense block, got {:?} after {previous:?}",
884                checkpoint.block_number
885            );
886        }
887        assert_eq!(checkpoint.block_number, Some(7), "one dense block cleared per run");
888
889        // With enough budget the remainder of the range completes in one run.
890        let (result, checkpoint) = run_prune(checkpoint, 1000);
891        assert!(result.progress.is_finished());
892        assert_eq!(checkpoint.block_number, Some(to_block));
893    }
894
895    /// Same guarantee for `prune_static_files`, which shares the changeset walk. `Segment::prune`
896    /// cannot reach it while `storage_v2` short-circuits to the `RocksDB` path, so it is called
897    /// directly.
898    #[test]
899    fn dense_block_advances_static_file_checkpoint() {
900        let db = TestStageDB::default();
901        let mut rng = generators::rng();
902
903        let blocks = random_block_range(
904            &mut rng,
905            0..=20,
906            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 0..1, ..Default::default() },
907        );
908        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
909
910        let accounts = random_eoa_accounts(&mut rng, 2).into_iter().collect::<BTreeMap<_, _>>();
911        let (changesets, _) = random_changeset_range(
912            &mut rng,
913            blocks.iter(),
914            accounts.into_iter().map(|(addr, acc)| (addr, (acc, Vec::new()))),
915            0..0,
916            0..0,
917        );
918        assert!(changesets.iter().all(|changeset| changeset.len() == 2));
919
920        // Changesets in static files. `StaticFileAccountChangesetWalker` resolves entries through
921        // `ChangeSetReader`, which only reads static files when `storage_v2` is set, so the walk
922        // needs v2 settings even though v2 routes `prune()` to the RocksDB path.
923        db.insert_changesets_to_static_files(changesets, None)
924            .expect("insert changesets to static files");
925        db.factory.set_storage_settings_cache(StorageSettings::v2());
926        assert!(db.table::<tables::AccountChangeSets>().unwrap().is_empty());
927
928        let to_block: BlockNumber = 15;
929        let prune_mode = PruneMode::Before(to_block);
930        let segment = AccountHistory::new(prune_mode);
931
932        let mut checkpoint = PruneCheckpoint { block_number: Some(4), tx_number: None, prune_mode };
933
934        for _ in 0..3 {
935            let previous = checkpoint.block_number;
936            let input = PruneInput {
937                previous_checkpoint: Some(checkpoint),
938                to_block,
939                // Halved internally by ACCOUNT_HISTORY_TABLES_TO_PRUNE, so 4 == one dense block.
940                limiter: PruneLimiter::default()
941                    .set_deleted_entries_limit(2 * ACCOUNT_HISTORY_TABLES_TO_PRUNE),
942            };
943            let range = input.get_next_block_range().unwrap();
944            let range_end = *range.end();
945
946            let provider = db.factory.database_provider_rw().unwrap();
947            provider.set_storage_settings_cache(StorageSettings::v2());
948            let result = segment.prune_static_files(&provider, input, range, range_end).unwrap();
949            segment
950                .save_checkpoint(
951                    &provider,
952                    result.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
953                )
954                .unwrap();
955            provider.commit().expect("commit");
956
957            checkpoint = db
958                .factory
959                .provider()
960                .unwrap()
961                .get_prune_checkpoint(PruneSegment::AccountHistory)
962                .unwrap()
963                .unwrap();
964
965            assert!(!result.progress.is_finished());
966            assert!(
967                checkpoint.block_number > previous,
968                "checkpoint must advance past the dense block, got {:?} after {previous:?}",
969                checkpoint.block_number
970            );
971        }
972        assert_eq!(checkpoint.block_number, Some(7), "one dense block cleared per run");
973    }
974}