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
22const 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 if provider.cached_storage_settings().storage_v2 {
77 return self.prune_rocksdb(provider, input, range, range_end);
78 }
79
80 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 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 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 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 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 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 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 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 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 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 let walker = StaticFileAccountChangesetWalker::new(provider, range);
280 for result in walker {
281 let (block_number, changeset) = result?;
282 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 let mut deleted_shards = 0usize;
303 let mut updated_shards = 0usize;
304
305 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 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 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 #[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 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 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 let changesets: Vec<ChangeSet> = vec![
660 vec![(addr1, account.clone(), vec![])], vec![(addr1, account.clone(), vec![])], vec![(addr1, account.clone(), vec![])], vec![(addr1, account.clone(), vec![])], vec![(addr1, account.clone(), vec![])], 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![])], ];
674
675 db.insert_changesets(changesets.clone(), None).expect("insert changesets");
676 db.insert_history(changesets.clone(), None).expect("insert history");
677
678 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 let deleted_entries_limit = 14; 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 assert!(!result.progress.is_finished(), "Expected HasMoreData since we stopped mid-block");
704
705 segment
707 .save_checkpoint(&provider, result.checkpoint.unwrap().as_prune_checkpoint(prune_mode))
708 .unwrap();
709 provider.commit().expect("commit");
710
711 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 let remaining_changesets = db.table::<tables::AccountChangeSets>().unwrap();
728 assert!(
732 !remaining_changesets.is_empty(),
733 "Should have remaining changesets for blocks 5-6"
734 );
735
736 let history = db.table::<tables::AccountsHistory>().unwrap();
739 for (key, _blocks) in &history {
740 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 let input2 = PruneInput {
751 previous_checkpoint: Some(checkpoint),
752 to_block: 10,
753 limiter: PruneLimiter::default().set_deleted_entries_limit(100), };
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 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 assert_eq!(final_checkpoint.block_number, Some(6), "Final checkpoint should be at block 6");
781
782 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 #[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 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 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 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 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 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 #[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 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 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}