1use super::metrics::{RocksDBMetrics, RocksDBOperation, ROCKSDB_TABLES};
2use crate::providers::{compute_history_rank, needs_prev_shard_check, HistoryInfo};
3use alloy_consensus::transaction::TxHashRef;
4use alloy_primitives::{
5 map::{AddressMap, HashMap},
6 Address, BlockNumber, TxNumber, B256,
7};
8use itertools::Itertools;
9use metrics::Label;
10use parking_lot::Mutex;
11use rayon::prelude::*;
12use reth_chain_state::ExecutedBlock;
13use reth_db_api::{
14 database_metrics::DatabaseMetrics,
15 models::{
16 sharded_key::NUM_OF_INDICES_IN_SHARD, storage_sharded_key::StorageShardedKey, ShardedKey,
17 StorageSettings,
18 },
19 table::{Compress, Decode, Decompress, Encode, Table},
20 tables, BlockNumberList, DatabaseError,
21};
22use reth_primitives_traits::{BlockBody as _, FastInstant as Instant};
23use reth_prune_types::PruneMode;
24use reth_storage_errors::{
25 db::{DatabaseErrorInfo, DatabaseWriteError, DatabaseWriteOperation, LogLevel},
26 provider::{ProviderError, ProviderResult},
27};
28use rocksdb::{
29 statistics::StatsLevel, BlockBasedOptions, Cache, ColumnFamilyDescriptor, CompactionPri,
30 DBCompressionType, DBRawIteratorWithThreadMode, IteratorMode, OptimisticTransactionDB,
31 OptimisticTransactionOptions, Options, ReadOptions, SnapshotWithThreadMode, Transaction,
32 WriteBatchWithTransaction, WriteBufferManager, WriteOptions, DB, DEFAULT_COLUMN_FAMILY_NAME,
33};
34use std::{
35 collections::BTreeMap,
36 fmt,
37 path::{Path, PathBuf},
38 sync::Arc,
39};
40use tracing::instrument;
41
42fn synced_write_options() -> WriteOptions {
44 let mut opts = WriteOptions::default();
45 opts.set_sync(true);
46 opts
47}
48
49pub(crate) type PendingRocksDBBatches = Arc<Mutex<Vec<WriteBatchWithTransaction<true>>>>;
51
52type RawKVResult = Result<(Box<[u8]>, Box<[u8]>), rocksdb::Error>;
54
55#[derive(Debug, Clone)]
57pub struct RocksDBTableStats {
58 pub sst_size_bytes: u64,
60 pub memtable_size_bytes: u64,
62 pub name: String,
64 pub estimated_num_keys: u64,
66 pub estimated_size_bytes: u64,
68 pub pending_compaction_bytes: u64,
70}
71
72#[derive(Debug, Clone)]
76pub struct RocksDBStats {
77 pub tables: Vec<RocksDBTableStats>,
79 pub wal_size_bytes: u64,
83}
84
85#[derive(Clone)]
87pub(crate) struct RocksDBWriteCtx {
88 pub first_block_number: BlockNumber,
90 pub prune_tx_lookup: Option<PruneMode>,
92 pub storage_settings: StorageSettings,
94 pub pending_batches: PendingRocksDBBatches,
96}
97
98impl fmt::Debug for RocksDBWriteCtx {
99 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
100 f.debug_struct("RocksDBWriteCtx")
101 .field("first_block_number", &self.first_block_number)
102 .field("prune_tx_lookup", &self.prune_tx_lookup)
103 .field("storage_settings", &self.storage_settings)
104 .field("pending_batches", &"<pending batches>")
105 .finish()
106 }
107}
108
109const DEFAULT_CACHE_SIZE: usize = 128 << 20;
111
112const DEFAULT_BLOCK_SIZE: usize = 16 * 1024;
114
115const DEFAULT_MAX_BACKGROUND_JOBS: i32 = 6;
117
118const LIMITED_MAX_OPEN_FILES: i32 = 512;
124
125const KEEP_ALL_FILES_OPEN: i32 = -1;
127
128const HIGH_FILE_DESCRIPTOR_LIMIT: u64 = 128 * 1024;
134
135const DEFAULT_BYTES_PER_SYNC: u64 = 1_048_576;
137
138const DEFAULT_WRITE_BUFFER_SIZE: usize = 128 << 20;
144
145const DEFAULT_WRITE_BUFFER_MANAGER_SIZE: usize = 4 * 1024 * 1024 * 1024;
150
151const DEFAULT_COMPRESS_BUF_CAPACITY: usize = 4096;
155
156const DEFAULT_AUTO_COMMIT_THRESHOLD: usize = 512 * 1024 * 1024;
163
164const DEFAULT_BAL_MIN_BLOB_SIZE: u64 = 4 * 1024;
168
169const DEFAULT_BAL_BLOB_FILE_SIZE: u64 = 256 * 1024 * 1024;
171
172pub struct RocksDBBuilder {
174 path: PathBuf,
175 column_families: Vec<String>,
176 enable_metrics: bool,
177 enable_statistics: bool,
178 log_level: rocksdb::LogLevel,
179 block_cache: Cache,
180 read_only: bool,
181}
182
183impl fmt::Debug for RocksDBBuilder {
184 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
185 f.debug_struct("RocksDBBuilder")
186 .field("path", &self.path)
187 .field("column_families", &self.column_families)
188 .field("enable_metrics", &self.enable_metrics)
189 .finish()
190 }
191}
192
193impl RocksDBBuilder {
194 pub fn new(path: impl AsRef<Path>) -> Self {
196 let cache = Cache::new_lru_cache(DEFAULT_CACHE_SIZE);
197 Self {
198 path: path.as_ref().to_path_buf(),
199 column_families: Vec::new(),
200 enable_metrics: false,
201 enable_statistics: false,
202 log_level: rocksdb::LogLevel::Info,
203 block_cache: cache,
204 read_only: false,
205 }
206 }
207
208 fn default_table_options(cache: &Cache) -> BlockBasedOptions {
210 let mut table_options = BlockBasedOptions::default();
211 table_options.set_block_size(DEFAULT_BLOCK_SIZE);
212 table_options.set_cache_index_and_filter_blocks(true);
213 table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
214 table_options.set_block_cache(cache);
216 table_options
217 }
218
219 fn default_options(
221 log_level: rocksdb::LogLevel,
222 cache: &Cache,
223 enable_statistics: bool,
224 ) -> Options {
225 let table_options = Self::default_table_options(cache);
227
228 let mut options = Options::default();
229 options.set_block_based_table_factory(&table_options);
230 options.create_if_missing(true);
231 options.create_missing_column_families(true);
232 options.set_max_background_jobs(DEFAULT_MAX_BACKGROUND_JOBS);
233 options.set_bytes_per_sync(DEFAULT_BYTES_PER_SYNC);
234 let write_buffer_manager =
235 WriteBufferManager::new_write_buffer_manager(DEFAULT_WRITE_BUFFER_MANAGER_SIZE, true);
236 options.set_write_buffer_manager(&write_buffer_manager);
237
238 options.set_bottommost_compression_type(DBCompressionType::Zstd);
239 options.set_bottommost_zstd_max_train_bytes(0, true);
240 options.set_compression_type(DBCompressionType::Lz4);
241 options.set_compaction_pri(CompactionPri::MinOverlappingRatio);
242
243 options.set_log_level(log_level);
244
245 options.set_max_open_files(select_max_open_files());
246
247 options.set_wal_ttl_seconds(0);
250 options.set_wal_size_limit_mb(0);
251
252 if enable_statistics {
254 options.enable_statistics();
255 options.set_statistics_level(StatsLevel::ExceptHistogramOrTimers);
265 }
266
267 options
268 }
269
270 fn default_column_family_options(cache: &Cache) -> Options {
272 let table_options = Self::default_table_options(cache);
274
275 let mut cf_options = Options::default();
276 cf_options.set_block_based_table_factory(&table_options);
277 cf_options.set_level_compaction_dynamic_level_bytes(true);
278 cf_options.set_compression_type(DBCompressionType::Lz4);
280 cf_options.set_bottommost_compression_type(DBCompressionType::Zstd);
281 cf_options.set_bottommost_zstd_max_train_bytes(0, true);
283 cf_options.set_write_buffer_size(DEFAULT_WRITE_BUFFER_SIZE);
284
285 cf_options
286 }
287
288 fn block_access_lists_column_family_options(cache: &Cache) -> Options {
290 let mut cf_options = Self::default_column_family_options(cache);
291 cf_options.set_enable_blob_files(true);
292 cf_options.set_min_blob_size(DEFAULT_BAL_MIN_BLOB_SIZE);
293 cf_options.set_blob_file_size(DEFAULT_BAL_BLOB_FILE_SIZE);
294 cf_options.set_blob_compression_type(DBCompressionType::Lz4);
295 cf_options
296 }
297
298 fn tx_hash_numbers_column_family_options(cache: &Cache) -> Options {
305 let mut table_options = BlockBasedOptions::default();
306 table_options.set_block_size(DEFAULT_BLOCK_SIZE);
307 table_options.set_cache_index_and_filter_blocks(true);
308 table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
309 table_options.set_block_cache(cache);
310 let mut cf_options = Options::default();
314 cf_options.set_block_based_table_factory(&table_options);
315 cf_options.set_level_compaction_dynamic_level_bytes(true);
316 cf_options.set_compression_type(DBCompressionType::None);
319 cf_options.set_bottommost_compression_type(DBCompressionType::None);
320
321 cf_options
322 }
323
324 pub fn with_table<T: Table>(mut self) -> Self {
326 self.column_families.push(T::NAME.to_string());
327 self
328 }
329
330 pub fn with_default_tables(self) -> Self {
339 self.with_table::<tables::TransactionHashNumbers>()
340 .with_table::<tables::AccountsHistory>()
341 .with_table::<tables::StoragesHistory>()
342 .with_table::<tables::BlockAccessLists>()
343 .with_table::<tables::BlockAccessListBlockNumbers>()
344 }
345
346 pub const fn with_metrics(mut self) -> Self {
348 self.enable_metrics = true;
349 self
350 }
351
352 pub const fn with_statistics(mut self) -> Self {
354 self.enable_statistics = true;
355 self
356 }
357
358 pub const fn with_database_log_level(mut self, log_level: Option<LogLevel>) -> Self {
360 if let Some(level) = log_level {
361 self.log_level = convert_log_level(level);
362 }
363 self
364 }
365
366 pub fn with_block_cache_size(mut self, capacity_bytes: usize) -> Self {
368 self.block_cache = Cache::new_lru_cache(capacity_bytes);
369 self
370 }
371
372 pub fn with_block_cache_size_opt(self, capacity_bytes: Option<usize>) -> Self {
374 if let Some(capacity_bytes) = capacity_bytes {
375 self.with_block_cache_size(capacity_bytes)
376 } else {
377 self
378 }
379 }
380
381 pub const fn with_read_only(mut self, read_only: bool) -> Self {
389 self.read_only = read_only;
390 self
391 }
392
393 pub fn build(self) -> ProviderResult<RocksDBProvider> {
395 let options =
396 Self::default_options(self.log_level, &self.block_cache, self.enable_statistics);
397
398 let mut cf_descriptors: Vec<ColumnFamilyDescriptor> = self
399 .column_families
400 .iter()
401 .map(|name| {
402 let cf_options = if name == tables::TransactionHashNumbers::NAME {
403 Self::tx_hash_numbers_column_family_options(&self.block_cache)
404 } else if name == tables::BlockAccessLists::NAME {
405 Self::block_access_lists_column_family_options(&self.block_cache)
406 } else {
407 Self::default_column_family_options(&self.block_cache)
408 };
409 ColumnFamilyDescriptor::new(name.clone(), cf_options)
410 })
411 .collect();
412
413 if RocksDBProvider::exists(&self.path) {
416 let existing_column_families = DB::list_cf(&options, &self.path).map_err(|e| {
417 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
418 message: e.to_string().into(),
419 code: -1,
420 }))
421 })?;
422 if self.read_only {
423 cf_descriptors.retain(|cf| {
426 !matches!(
427 cf.name(),
428 tables::BlockAccessLists::NAME | tables::BlockAccessListBlockNumbers::NAME
429 ) || existing_column_families.iter().any(|name| name == cf.name())
430 });
431 }
432 let unknown_column_families: Vec<String> = existing_column_families
433 .into_iter()
434 .filter(|name| {
435 name != DEFAULT_COLUMN_FAMILY_NAME && !self.column_families.contains(name)
436 })
437 .collect();
438 if !unknown_column_families.is_empty() {
439 tracing::debug!(
440 target: "providers::rocksdb",
441 column_families = ?unknown_column_families,
442 "Preserving unknown column families"
443 );
444 cf_descriptors.extend(unknown_column_families.into_iter().map(|name| {
445 ColumnFamilyDescriptor::new(
446 name,
447 Self::default_column_family_options(&self.block_cache),
448 )
449 }));
450 }
451 }
452
453 let metrics = self.enable_metrics.then(RocksDBMetrics::default);
454
455 if self.read_only {
456 let mut options = options;
459 options.set_max_open_files(KEEP_ALL_FILES_OPEN);
460
461 let secondary_path = self
462 .path
463 .parent()
464 .unwrap_or(&self.path)
465 .join(format!("rocksdb-secondary-tmp-{}", std::process::id()));
466 reth_fs_util::create_dir_all(&secondary_path).map_err(ProviderError::other)?;
467
468 let db = DB::open_cf_descriptors_as_secondary(
469 &options,
470 &self.path,
471 &secondary_path,
472 cf_descriptors,
473 )
474 .map_err(|e| {
475 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
476 message: e.to_string().into(),
477 code: -1,
478 }))
479 })?;
480 Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::Secondary {
481 db,
482 metrics,
483 secondary_path,
484 })))
485 } else {
486 let db =
491 OptimisticTransactionDB::open_cf_descriptors(&options, &self.path, cf_descriptors)
492 .map_err(|e| {
493 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
494 message: e.to_string().into(),
495 code: -1,
496 }))
497 })?;
498 Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::ReadWrite { db, metrics })))
499 }
500 }
501}
502
503macro_rules! compress_to_buf_or_ref {
506 ($buf:expr, $value:expr) => {
507 if let Some(value) = $value.uncompressable_ref() {
508 Some(value)
509 } else {
510 $buf.clear();
511 $value.compress_to_buf(&mut $buf);
512 None
513 }
514 };
515}
516
517#[derive(Debug)]
519pub struct RocksDBProvider(Arc<RocksDBProviderInner>);
520
521enum RocksDBProviderInner {
523 ReadWrite {
525 db: OptimisticTransactionDB,
527 metrics: Option<RocksDBMetrics>,
529 },
530 Secondary {
534 db: DB,
536 metrics: Option<RocksDBMetrics>,
538 secondary_path: PathBuf,
540 },
541}
542
543impl RocksDBProviderInner {
544 const fn metrics(&self) -> Option<&RocksDBMetrics> {
546 match self {
547 Self::ReadWrite { metrics, .. } | Self::Secondary { metrics, .. } => metrics.as_ref(),
548 }
549 }
550
551 fn db_rw(&self) -> &OptimisticTransactionDB {
553 match self {
554 Self::ReadWrite { db, .. } => db,
555 Self::Secondary { .. } => {
556 panic!("Cannot perform write operation on secondary RocksDB provider")
557 }
558 }
559 }
560
561 fn cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
563 let cf = match self {
564 Self::ReadWrite { db, .. } => db.cf_handle(T::NAME),
565 Self::Secondary { db, .. } => db.cf_handle(T::NAME),
566 };
567 cf.ok_or_else(|| DatabaseError::Other(format!("Column family '{}' not found", T::NAME)))
568 }
569
570 fn get_cf(
572 &self,
573 cf: &rocksdb::ColumnFamily,
574 key: impl AsRef<[u8]>,
575 ) -> Result<Option<Vec<u8>>, rocksdb::Error> {
576 match self {
577 Self::ReadWrite { db, .. } => db.get_cf(cf, key),
578 Self::Secondary { db, .. } => db.get_cf(cf, key),
579 }
580 }
581
582 fn put_cf(
584 &self,
585 cf: &rocksdb::ColumnFamily,
586 key: impl AsRef<[u8]>,
587 value: impl AsRef<[u8]>,
588 ) -> Result<(), rocksdb::Error> {
589 self.db_rw().put_cf(cf, key, value)
590 }
591
592 fn delete_cf(
594 &self,
595 cf: &rocksdb::ColumnFamily,
596 key: impl AsRef<[u8]>,
597 ) -> Result<(), rocksdb::Error> {
598 self.db_rw().delete_cf(cf, key)
599 }
600
601 fn delete_range_cf<K: AsRef<[u8]>>(
603 &self,
604 cf: &rocksdb::ColumnFamily,
605 from: K,
606 to: K,
607 ) -> Result<(), rocksdb::Error> {
608 self.db_rw().delete_range_cf(cf, from, to)
609 }
610
611 fn iterator_cf(
613 &self,
614 cf: &rocksdb::ColumnFamily,
615 mode: IteratorMode<'_>,
616 ) -> RocksDBIterEnum<'_> {
617 match self {
618 Self::ReadWrite { db, .. } => RocksDBIterEnum::ReadWrite(db.iterator_cf(cf, mode)),
619 Self::Secondary { db, .. } => RocksDBIterEnum::ReadOnly(db.iterator_cf(cf, mode)),
620 }
621 }
622
623 fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
628 match self {
629 Self::ReadWrite { db, .. } => RocksDBRawIterEnum::ReadWrite(db.raw_iterator_cf(cf)),
630 Self::Secondary { db, .. } => RocksDBRawIterEnum::ReadOnly(db.raw_iterator_cf(cf)),
631 }
632 }
633
634 fn iterator_cf_inclusive_range(
640 &self,
641 cf: &rocksdb::ColumnFamily,
642 from: &[u8],
643 last: &[u8],
644 ) -> RocksDBIterEnum<'_> {
645 let mut upper_bound = Vec::with_capacity(last.len() + 1);
647 upper_bound.extend_from_slice(last);
648 upper_bound.push(0);
649
650 let mut opts = ReadOptions::default();
651 opts.set_iterate_upper_bound(upper_bound);
652 let mode = IteratorMode::From(from, rocksdb::Direction::Forward);
653 match self {
654 Self::ReadWrite { db, .. } => {
655 RocksDBIterEnum::ReadWrite(db.iterator_cf_opt(cf, opts, mode))
656 }
657 Self::Secondary { db, .. } => {
658 RocksDBIterEnum::ReadOnly(db.iterator_cf_opt(cf, opts, mode))
659 }
660 }
661 }
662
663 fn snapshot(&self) -> RocksReadSnapshotInner<'_> {
665 match self {
666 Self::ReadWrite { db, .. } => RocksReadSnapshotInner::ReadWrite(db.snapshot()),
667 Self::Secondary { db, .. } => RocksReadSnapshotInner::Secondary(db),
668 }
669 }
670
671 fn path(&self) -> &Path {
673 match self {
674 Self::ReadWrite { db, .. } => db.path(),
675 Self::Secondary { db, .. } => db.path(),
676 }
677 }
678
679 fn wal_size_bytes(&self) -> u64 {
683 let path = self.path();
684
685 match std::fs::read_dir(path) {
686 Ok(entries) => entries
687 .filter_map(|e| e.ok())
688 .filter(|e| e.path().extension().is_some_and(|ext| ext == "log"))
689 .filter_map(|e| e.metadata().ok())
690 .map(|m| m.len())
691 .sum(),
692 Err(_) => 0,
693 }
694 }
695
696 fn table_stats(&self) -> Vec<RocksDBTableStats> {
698 let mut stats = Vec::new();
699
700 macro_rules! collect_stats {
701 ($db:expr) => {
702 for cf_name in ROCKSDB_TABLES {
703 if let Some(cf) = $db.cf_handle(cf_name) {
704 let estimated_num_keys = $db
705 .property_int_value_cf(cf, rocksdb::properties::ESTIMATE_NUM_KEYS)
706 .ok()
707 .flatten()
708 .unwrap_or(0);
709
710 let sst_size = $db
712 .property_int_value_cf(cf, rocksdb::properties::LIVE_SST_FILES_SIZE)
713 .ok()
714 .flatten()
715 .unwrap_or(0);
716
717 let memtable_size = $db
718 .property_int_value_cf(cf, rocksdb::properties::SIZE_ALL_MEM_TABLES)
719 .ok()
720 .flatten()
721 .unwrap_or(0);
722
723 let estimated_size_bytes = sst_size + memtable_size;
724
725 let pending_compaction_bytes = $db
726 .property_int_value_cf(
727 cf,
728 rocksdb::properties::ESTIMATE_PENDING_COMPACTION_BYTES,
729 )
730 .ok()
731 .flatten()
732 .unwrap_or(0);
733
734 stats.push(RocksDBTableStats {
735 sst_size_bytes: sst_size,
736 memtable_size_bytes: memtable_size,
737 name: cf_name.to_string(),
738 estimated_num_keys,
739 estimated_size_bytes,
740 pending_compaction_bytes,
741 });
742 }
743 }
744 };
745 }
746
747 match self {
748 Self::ReadWrite { db, .. } => collect_stats!(db),
749 Self::Secondary { db, .. } => collect_stats!(db),
750 }
751
752 stats
753 }
754
755 fn db_stats(&self) -> RocksDBStats {
757 RocksDBStats { tables: self.table_stats(), wal_size_bytes: self.wal_size_bytes() }
758 }
759}
760
761impl fmt::Debug for RocksDBProviderInner {
762 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
763 match self {
764 Self::ReadWrite { metrics, .. } => f
765 .debug_struct("RocksDBProviderInner::ReadWrite")
766 .field("db", &"<OptimisticTransactionDB>")
767 .field("metrics", metrics)
768 .finish(),
769 Self::Secondary { metrics, .. } => f
770 .debug_struct("RocksDBProviderInner::Secondary")
771 .field("db", &"<DB (secondary)>")
772 .field("metrics", metrics)
773 .finish(),
774 }
775 }
776}
777
778impl Drop for RocksDBProviderInner {
779 fn drop(&mut self) {
780 match self {
781 Self::ReadWrite { db, .. } => {
782 if let Err(e) = db.flush_wal(true) {
785 tracing::warn!(target: "providers::rocksdb", ?e, "Failed to flush WAL on drop");
786 }
787 for cf_name in ROCKSDB_TABLES {
788 if let Some(cf) = db.cf_handle(cf_name) &&
789 let Err(e) = db.flush_cf(&cf)
790 {
791 tracing::warn!(target: "providers::rocksdb", cf = cf_name, ?e, "Failed to flush CF on drop");
792 }
793 }
794 db.cancel_all_background_work(true);
795 }
796 Self::Secondary { db, secondary_path, .. } => {
797 db.cancel_all_background_work(true);
798 let _ = std::fs::remove_dir_all(secondary_path);
799 }
800 }
801 }
802}
803
804impl Clone for RocksDBProvider {
805 fn clone(&self) -> Self {
806 Self(self.0.clone())
807 }
808}
809
810impl DatabaseMetrics for RocksDBProvider {
811 fn gauge_metrics(&self) -> Vec<(&'static str, f64, Vec<Label>)> {
812 let mut metrics = Vec::new();
813
814 for stat in self.table_stats() {
815 metrics.push((
816 "rocksdb.table_size",
817 stat.estimated_size_bytes as f64,
818 vec![Label::new("table", stat.name.clone())],
819 ));
820 metrics.push((
821 "rocksdb.table_entries",
822 stat.estimated_num_keys as f64,
823 vec![Label::new("table", stat.name.clone())],
824 ));
825 metrics.push((
826 "rocksdb.pending_compaction_bytes",
827 stat.pending_compaction_bytes as f64,
828 vec![Label::new("table", stat.name.clone())],
829 ));
830 metrics.push((
831 "rocksdb.sst_size",
832 stat.sst_size_bytes as f64,
833 vec![Label::new("table", stat.name.clone())],
834 ));
835 metrics.push((
836 "rocksdb.memtable_size",
837 stat.memtable_size_bytes as f64,
838 vec![Label::new("table", stat.name)],
839 ));
840 }
841
842 metrics.push(("rocksdb.wal_size", self.wal_size_bytes() as f64, vec![]));
844
845 metrics
846 }
847}
848
849impl RocksDBProvider {
850 pub fn new(path: impl AsRef<Path>) -> ProviderResult<Self> {
852 RocksDBBuilder::new(path).build()
853 }
854
855 pub fn builder(path: impl AsRef<Path>) -> RocksDBBuilder {
857 RocksDBBuilder::new(path)
858 }
859
860 pub fn exists(path: impl AsRef<Path>) -> bool {
865 path.as_ref().join("CURRENT").exists()
866 }
867
868 pub fn is_read_only(&self) -> bool {
870 matches!(self.0.as_ref(), RocksDBProviderInner::Secondary { .. })
871 }
872
873 pub fn try_catch_up_with_primary(&self) -> ProviderResult<()> {
878 match self.0.as_ref() {
879 RocksDBProviderInner::Secondary { db, .. } => {
880 db.try_catch_up_with_primary().map_err(|e| {
881 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
882 message: e.to_string().into(),
883 code: -1,
884 }))
885 })
886 }
887 _ => Ok(()),
888 }
889 }
890
891 pub fn snapshot(&self) -> RocksReadSnapshot<'_> {
895 RocksReadSnapshot {
896 accounts_history_iter: Mutex::new(None),
897 storages_history_iter: Mutex::new(None),
898 inner: self.0.snapshot(),
899 provider: &self.0,
900 }
901 }
902
903 pub fn owned_snapshot(&self) -> OwnedRocksReadSnapshot {
908 OwnedRocksReadSnapshot::new(self)
909 }
910
911 pub fn tx(&self) -> RocksTx<'_> {
919 let write_options = synced_write_options();
920 let txn_options = OptimisticTransactionOptions::default();
921 let inner = self.0.db_rw().transaction_opt(&write_options, &txn_options);
922 RocksTx { inner, provider: self }
923 }
924
925 pub fn batch(&self) -> RocksDBBatch<'_> {
933 RocksDBBatch {
934 provider: self,
935 inner: WriteBatchWithTransaction::<true>::default(),
936 buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
937 auto_commit_threshold: None,
938 }
939 }
940
941 pub fn batch_with_auto_commit(&self) -> RocksDBBatch<'_> {
947 RocksDBBatch {
948 provider: self,
949 inner: WriteBatchWithTransaction::<true>::default(),
950 buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
951 auto_commit_threshold: Some(DEFAULT_AUTO_COMMIT_THRESHOLD),
952 }
953 }
954
955 fn get_cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
957 self.0.cf_handle::<T>()
958 }
959
960 pub(crate) fn has_table<T: Table>(&self) -> bool {
962 match self.0.as_ref() {
963 RocksDBProviderInner::ReadWrite { db, .. } => db.cf_handle(T::NAME).is_some(),
964 RocksDBProviderInner::Secondary { db, .. } => db.cf_handle(T::NAME).is_some(),
965 }
966 }
967
968 fn execute_with_operation_metric<R>(
970 &self,
971 operation: RocksDBOperation,
972 table: &'static str,
973 f: impl FnOnce(&Self) -> R,
974 ) -> R {
975 let start = self.0.metrics().map(|_| Instant::now());
976 let res = f(self);
977
978 if let (Some(start), Some(metrics)) = (start, self.0.metrics()) {
979 metrics.record_operation(operation, table, start.elapsed());
980 }
981
982 res
983 }
984
985 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
987 self.get_encoded::<T>(&key.encode())
988 }
989
990 pub fn get_encoded<T: Table>(
992 &self,
993 key: &<T::Key as Encode>::Encoded,
994 ) -> ProviderResult<Option<T::Value>> {
995 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
996 let result = this.0.get_cf(this.get_cf_handle::<T>()?, key.as_ref()).map_err(|e| {
997 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
998 message: e.to_string().into(),
999 code: -1,
1000 }))
1001 })?;
1002
1003 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
1004 })
1005 }
1006
1007 pub fn get_raw<T: Table>(&self, key: T::Key) -> ProviderResult<Option<Vec<u8>>> {
1009 let encoded = key.encode();
1010 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
1011 this.0.get_cf(this.get_cf_handle::<T>()?, encoded.as_ref()).map_err(|e| {
1012 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1013 message: e.to_string().into(),
1014 code: -1,
1015 }))
1016 })
1017 })
1018 }
1019
1020 pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
1025 let encoded_key = key.encode();
1026 self.put_encoded::<T>(&encoded_key, value)
1027 }
1028
1029 pub fn put_encoded<T: Table>(
1034 &self,
1035 key: &<T::Key as Encode>::Encoded,
1036 value: &T::Value,
1037 ) -> ProviderResult<()> {
1038 self.execute_with_operation_metric(RocksDBOperation::Put, T::NAME, |this| {
1039 let mut buf = Vec::new();
1043 let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
1044
1045 this.0.put_cf(this.get_cf_handle::<T>()?, key, value_bytes).map_err(|e| {
1046 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1047 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1048 operation: DatabaseWriteOperation::PutUpsert,
1049 table_name: T::NAME,
1050 key: key.as_ref().to_vec(),
1051 })))
1052 })
1053 })
1054 }
1055
1056 pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
1061 self.execute_with_operation_metric(RocksDBOperation::Delete, T::NAME, |this| {
1062 this.0.delete_cf(this.get_cf_handle::<T>()?, key.encode().as_ref()).map_err(|e| {
1063 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
1064 message: e.to_string().into(),
1065 code: -1,
1066 }))
1067 })
1068 })
1069 }
1070
1071 pub fn clear<T: Table>(&self) -> ProviderResult<()> {
1077 let cf = self.get_cf_handle::<T>()?;
1078
1079 self.0.delete_range_cf(cf, &[] as &[u8], &[0xFF; 256]).map_err(|e| {
1080 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
1081 message: e.to_string().into(),
1082 code: -1,
1083 }))
1084 })?;
1085
1086 Ok(())
1087 }
1088
1089 fn get_boundary<T: Table>(
1091 &self,
1092 mode: IteratorMode<'_>,
1093 ) -> ProviderResult<Option<(T::Key, T::Value)>> {
1094 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
1095 let cf = this.get_cf_handle::<T>()?;
1096 let mut iter = this.0.iterator_cf(cf, mode);
1097
1098 match iter.next() {
1099 Some(Ok((key_bytes, value_bytes))) => {
1100 let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
1101 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1102 let value = T::Value::decompress(&value_bytes)
1103 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1104 Ok(Some((key, value)))
1105 }
1106 Some(Err(e)) => {
1107 Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1108 message: e.to_string().into(),
1109 code: -1,
1110 })))
1111 }
1112 None => Ok(None),
1113 }
1114 })
1115 }
1116
1117 #[inline]
1119 pub fn first<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1120 self.get_boundary::<T>(IteratorMode::Start)
1121 }
1122
1123 #[inline]
1125 pub fn last<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1126 self.get_boundary::<T>(IteratorMode::End)
1127 }
1128
1129 pub fn iter<T: Table>(&self) -> ProviderResult<RocksDBIter<'_, T>> {
1133 let cf = self.get_cf_handle::<T>()?;
1134 let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1135 Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1136 }
1137
1138 pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksDBIter<'_, T>> {
1142 let cf = self.get_cf_handle::<T>()?;
1143 let encoded_key = key.encode();
1144 let iter = self
1145 .0
1146 .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
1147 Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1148 }
1149
1150 pub fn table_stats(&self) -> Vec<RocksDBTableStats> {
1154 self.0.table_stats()
1155 }
1156
1157 pub fn wal_size_bytes(&self) -> u64 {
1163 self.0.wal_size_bytes()
1164 }
1165
1166 pub fn db_stats(&self) -> RocksDBStats {
1170 self.0.db_stats()
1171 }
1172
1173 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(tables = ?tables))]
1184 pub fn flush(&self, tables: &[&'static str]) -> ProviderResult<()> {
1185 let db = self.0.db_rw();
1186
1187 for cf_name in tables {
1188 if let Some(cf) = db.cf_handle(cf_name) {
1189 db.flush_cf(&cf).map_err(|e| {
1190 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1191 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1192 operation: DatabaseWriteOperation::Flush,
1193 table_name: cf_name,
1194 key: Vec::new(),
1195 })))
1196 })?;
1197 }
1198 }
1199
1200 db.flush_wal(true).map_err(|e| {
1201 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1202 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1203 operation: DatabaseWriteOperation::Flush,
1204 table_name: "WAL",
1205 key: Vec::new(),
1206 })))
1207 })?;
1208
1209 Ok(())
1210 }
1211
1212 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1224 pub fn flush_and_compact(&self) -> ProviderResult<()> {
1225 self.flush(ROCKSDB_TABLES)?;
1226
1227 let db = self.0.db_rw();
1228
1229 for cf_name in ROCKSDB_TABLES {
1230 if let Some(cf) = db.cf_handle(cf_name) {
1231 db.compact_range_cf(&cf, None::<&[u8]>, None::<&[u8]>);
1232 }
1233 }
1234
1235 Ok(())
1236 }
1237
1238 pub fn raw_iter<T: Table>(&self) -> ProviderResult<RocksDBRawIter<'_>> {
1242 let cf = self.get_cf_handle::<T>()?;
1243 let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1244 Ok(RocksDBRawIter { inner: iter })
1245 }
1246
1247 pub(crate) fn raw_key_iter_from<T: Table>(
1249 &self,
1250 key: T::Key,
1251 ) -> ProviderResult<RocksDBRawKeyIter<'_>> {
1252 let cf = self.get_cf_handle::<T>()?;
1253 let encoded_key = key.encode();
1254 let mut iter = self.0.raw_iterator_cf(cf);
1255 iter.seek(encoded_key.as_ref());
1256 Ok(RocksDBRawKeyIter { inner: iter })
1257 }
1258
1259 pub fn account_history_shards(
1264 &self,
1265 address: Address,
1266 ) -> ProviderResult<Vec<(ShardedKey<Address>, BlockNumberList)>> {
1267 let cf = self.get_cf_handle::<tables::AccountsHistory>()?;
1269
1270 let start_key = ShardedKey::new(address, 0u64);
1273 let start_bytes = start_key.encode();
1274 let last_bytes = ShardedKey::new(address, u64::MAX).encode();
1276
1277 let iter =
1280 self.0.iterator_cf_inclusive_range(cf, start_bytes.as_ref(), last_bytes.as_ref());
1281
1282 let mut result = Vec::new();
1283 for item in iter {
1284 match item {
1285 Ok((key_bytes, value_bytes)) => {
1286 let key = ShardedKey::<Address>::decode(&key_bytes)
1288 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1289
1290 if key.key != address {
1292 break;
1293 }
1294
1295 let value = BlockNumberList::decompress(&value_bytes)
1297 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1298
1299 result.push((key, value));
1300 }
1301 Err(e) => {
1302 return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1303 message: e.to_string().into(),
1304 code: -1,
1305 })));
1306 }
1307 }
1308 }
1309
1310 Ok(result)
1311 }
1312
1313 pub fn storage_history_shards(
1318 &self,
1319 address: Address,
1320 storage_key: B256,
1321 ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1322 let cf = self.get_cf_handle::<tables::StoragesHistory>()?;
1323
1324 let start_key = StorageShardedKey::new(address, storage_key, 0u64);
1325 let start_bytes = start_key.encode();
1326 let last_bytes = StorageShardedKey::new(address, storage_key, u64::MAX).encode();
1327
1328 let iter =
1330 self.0.iterator_cf_inclusive_range(cf, start_bytes.as_ref(), last_bytes.as_ref());
1331
1332 let mut result = Vec::new();
1333 for item in iter {
1334 match item {
1335 Ok((key_bytes, value_bytes)) => {
1336 let key = StorageShardedKey::decode(&key_bytes)
1337 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1338
1339 if key.address != address || key.sharded_key.key != storage_key {
1340 break;
1341 }
1342
1343 let value = BlockNumberList::decompress(&value_bytes)
1344 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1345
1346 result.push((key, value));
1347 }
1348 Err(e) => {
1349 return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1350 message: e.to_string().into(),
1351 code: -1,
1352 })));
1353 }
1354 }
1355 }
1356
1357 Ok(result)
1358 }
1359
1360 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1368 pub fn unwind_account_history_indices(
1369 &self,
1370 last_indices: &[(Address, BlockNumber)],
1371 ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1372 let mut address_min_block: AddressMap<BlockNumber> =
1373 AddressMap::with_capacity_and_hasher(last_indices.len(), Default::default());
1374 for &(address, block_number) in last_indices {
1375 address_min_block
1376 .entry(address)
1377 .and_modify(|min| *min = (*min).min(block_number))
1378 .or_insert(block_number);
1379 }
1380
1381 let mut batch = self.batch();
1382 for (address, min_block) in address_min_block {
1383 match min_block.checked_sub(1) {
1384 Some(keep_to) => batch.unwind_account_history_to(address, keep_to)?,
1385 None => batch.clear_account_history(address)?,
1386 }
1387 }
1388
1389 Ok(batch.into_inner())
1390 }
1391
1392 pub fn unwind_storage_history_indices(
1400 &self,
1401 storage_changesets: &[(Address, B256, BlockNumber)],
1402 ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1403 let mut key_min_block: HashMap<(Address, B256), BlockNumber> =
1404 HashMap::with_capacity_and_hasher(storage_changesets.len(), Default::default());
1405 for &(address, storage_key, block_number) in storage_changesets {
1406 key_min_block
1407 .entry((address, storage_key))
1408 .and_modify(|min| *min = (*min).min(block_number))
1409 .or_insert(block_number);
1410 }
1411
1412 let mut batch = self.batch();
1413 for ((address, storage_key), min_block) in key_min_block {
1414 match min_block.checked_sub(1) {
1415 Some(keep_to) => batch.unwind_storage_history_to(address, storage_key, keep_to)?,
1416 None => batch.clear_storage_history(address, storage_key)?,
1417 }
1418 }
1419
1420 Ok(batch.into_inner())
1421 }
1422
1423 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1425 pub fn write_batch<F>(&self, f: F) -> ProviderResult<()>
1426 where
1427 F: FnOnce(&mut RocksDBBatch<'_>) -> ProviderResult<()>,
1428 {
1429 self.execute_with_operation_metric(RocksDBOperation::BatchWrite, "Batch", |this| {
1430 let mut batch_handle = this.batch();
1431 f(&mut batch_handle)?;
1432 batch_handle.commit()
1433 })
1434 }
1435
1436 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = batch.len(), batch_size = batch.size_in_bytes()))]
1444 pub fn commit_batch(&self, batch: WriteBatchWithTransaction<true>) -> ProviderResult<()> {
1445 self.0.db_rw().write_opt(batch, &synced_write_options()).map_err(|e| {
1446 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1447 message: e.to_string().into(),
1448 code: -1,
1449 }))
1450 })
1451 }
1452
1453 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(num_blocks = blocks.len(), first_block = ctx.first_block_number))]
1459 pub(crate) fn write_blocks_data<N: reth_node_types::NodePrimitives>(
1460 &self,
1461 blocks: &[ExecutedBlock<N>],
1462 tx_nums: &[TxNumber],
1463 ctx: RocksDBWriteCtx,
1464 runtime: &reth_tasks::Runtime,
1465 ) -> ProviderResult<()> {
1466 if !ctx.storage_settings.storage_v2 {
1467 return Ok(());
1468 }
1469
1470 let mut r_tx_hash = None;
1471 let mut r_account_history = None;
1472 let mut r_storage_history = None;
1473
1474 let write_tx_hash =
1475 ctx.storage_settings.storage_v2 && ctx.prune_tx_lookup.is_none_or(|m| !m.is_full());
1476 let write_account_history = ctx.storage_settings.storage_v2;
1477 let write_storage_history = ctx.storage_settings.storage_v2;
1478
1479 let span = tracing::Span::current();
1482 runtime.storage_pool().in_place_scope(|s| {
1483 if write_tx_hash {
1484 s.spawn(|_| {
1485 let _guard = span.enter();
1486 r_tx_hash = Some(self.write_tx_hash_numbers(blocks, tx_nums, &ctx));
1487 });
1488 }
1489
1490 if write_account_history {
1491 s.spawn(|_| {
1492 let _guard = span.enter();
1493 r_account_history = Some(self.write_account_history(blocks, &ctx));
1494 });
1495 }
1496
1497 if write_storage_history {
1498 s.spawn(|_| {
1499 let _guard = span.enter();
1500 r_storage_history = Some(self.write_storage_history(blocks, &ctx));
1501 });
1502 }
1503 });
1504
1505 if write_tx_hash {
1506 r_tx_hash.ok_or_else(|| {
1507 ProviderError::Database(DatabaseError::Other(
1508 "rocksdb tx-hash write thread panicked".into(),
1509 ))
1510 })??;
1511 }
1512 if write_account_history {
1513 r_account_history.ok_or_else(|| {
1514 ProviderError::Database(DatabaseError::Other(
1515 "rocksdb account-history write thread panicked".into(),
1516 ))
1517 })??;
1518 }
1519 if write_storage_history {
1520 r_storage_history.ok_or_else(|| {
1521 ProviderError::Database(DatabaseError::Other(
1522 "rocksdb storage-history write thread panicked".into(),
1523 ))
1524 })??;
1525 }
1526
1527 Ok(())
1528 }
1529
1530 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1532 fn write_tx_hash_numbers<N: reth_node_types::NodePrimitives>(
1533 &self,
1534 blocks: &[ExecutedBlock<N>],
1535 tx_nums: &[TxNumber],
1536 ctx: &RocksDBWriteCtx,
1537 ) -> ProviderResult<()> {
1538 let mut batch = self.batch();
1539 for (block, &first_tx_num) in blocks.iter().zip(tx_nums) {
1540 let body = block.recovered_block().body();
1541 for (tx_num, transaction) in (first_tx_num..).zip(body.transactions_iter()) {
1542 batch.put::<tables::TransactionHashNumbers>(*transaction.tx_hash(), &tx_num)?;
1543 }
1544 }
1545 ctx.pending_batches.lock().push(batch.into_inner());
1546 Ok(())
1547 }
1548
1549 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1553 fn write_account_history<N: reth_node_types::NodePrimitives>(
1554 &self,
1555 blocks: &[ExecutedBlock<N>],
1556 ctx: &RocksDBWriteCtx,
1557 ) -> ProviderResult<()> {
1558 let mut batch = self.batch();
1559 let mut account_history: BTreeMap<Address, Vec<u64>> = BTreeMap::new();
1560
1561 for (block_idx, block) in blocks.iter().enumerate() {
1562 let block_number = ctx.first_block_number + block_idx as u64;
1563 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1564
1565 for account_block_reverts in reverts.accounts {
1568 for (address, _) in account_block_reverts {
1569 account_history.entry(address).or_default().push(block_number);
1570 }
1571 }
1572 }
1573
1574 for (address, indices) in account_history {
1576 batch.append_account_history_shard(address, indices)?;
1577 }
1578 ctx.pending_batches.lock().push(batch.into_inner());
1579 Ok(())
1580 }
1581
1582 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1586 fn write_storage_history<N: reth_node_types::NodePrimitives>(
1587 &self,
1588 blocks: &[ExecutedBlock<N>],
1589 ctx: &RocksDBWriteCtx,
1590 ) -> ProviderResult<()> {
1591 let mut storage_history: BTreeMap<(Address, B256), Vec<u64>> = BTreeMap::new();
1592
1593 for (block_idx, block) in blocks.iter().enumerate() {
1594 let block_number = ctx.first_block_number + block_idx as u64;
1595 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1596
1597 for storage_block_reverts in reverts.storage {
1600 for revert in storage_block_reverts {
1601 for (slot, _) in revert.storage_revert {
1602 let plain_key = B256::from(slot);
1603 storage_history
1604 .entry((revert.address, plain_key))
1605 .or_default()
1606 .push(block_number);
1607 }
1608 }
1609 }
1610 }
1611
1612 let shard_puts = storage_history
1613 .into_par_iter()
1614 .map(|((address, slot), indices)| {
1615 self.storage_history_shards_to_put(address, slot, indices)
1616 })
1617 .collect::<ProviderResult<Vec<_>>>()?;
1618
1619 let mut batch = self.batch();
1620 for shards in shard_puts {
1621 for (key, shard) in shards {
1622 batch.put::<tables::StoragesHistory>(key, &shard)?;
1623 }
1624 }
1625 ctx.pending_batches.lock().push(batch.into_inner());
1626 Ok(())
1627 }
1628
1629 fn storage_history_shards_to_put(
1632 &self,
1633 address: Address,
1634 storage_key: B256,
1635 indices: Vec<u64>,
1636 ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1637 if indices.is_empty() {
1638 return Ok(Vec::new());
1639 }
1640
1641 debug_assert!(
1642 indices.windows(2).all(|w| w[0] < w[1]),
1643 "indices must be strictly increasing: {:?}",
1644 indices
1645 );
1646
1647 let last_key = StorageShardedKey::last(address, storage_key);
1648 let last_shard_opt = self.get::<tables::StoragesHistory>(last_key.clone())?;
1649 let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
1650
1651 last_shard.append(indices).map_err(ProviderError::other)?;
1652
1653 if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
1654 return Ok(vec![(last_key, last_shard)]);
1655 }
1656
1657 let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
1658 let mut chunks_peekable = chunks.into_iter().peekable();
1659 let mut shards = Vec::new();
1660
1661 while let Some(chunk) = chunks_peekable.next() {
1662 let shard = BlockNumberList::new_pre_sorted(chunk);
1663 let highest_block_number = if chunks_peekable.peek().is_some() {
1664 shard.iter().next_back().expect("`chunks` does not return empty list")
1665 } else {
1666 u64::MAX
1667 };
1668
1669 shards
1670 .push((StorageShardedKey::new(address, storage_key, highest_block_number), shard));
1671 }
1672
1673 Ok(shards)
1674 }
1675}
1676
1677pub struct RocksReadSnapshot<'db> {
1685 accounts_history_iter: Mutex<Option<RocksDBRawIterEnum<'db>>>,
1689 storages_history_iter: Mutex<Option<RocksDBRawIterEnum<'db>>>,
1693 inner: RocksReadSnapshotInner<'db>,
1694 provider: &'db RocksDBProviderInner,
1695}
1696
1697enum RocksReadSnapshotInner<'db> {
1699 ReadWrite(SnapshotWithThreadMode<'db, OptimisticTransactionDB>),
1701 Secondary(&'db DB),
1703}
1704
1705impl fmt::Debug for RocksReadSnapshot<'_> {
1706 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1707 f.debug_struct("RocksReadSnapshot")
1708 .field("provider", &self.provider)
1709 .finish_non_exhaustive()
1710 }
1711}
1712
1713impl<'db> RocksReadSnapshot<'db> {
1714 fn cf_handle<T: Table>(&self) -> Result<&'db rocksdb::ColumnFamily, DatabaseError> {
1716 self.provider.cf_handle::<T>()
1717 }
1718
1719 fn new_raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'db> {
1726 match &self.inner {
1727 RocksReadSnapshotInner::ReadWrite(snap) => {
1728 let mut readopts = ReadOptions::default();
1729 readopts.set_snapshot(snap);
1730 RocksDBRawIterEnum::ReadWrite(
1731 self.provider.db_rw().raw_iterator_cf_opt(cf, readopts),
1732 )
1733 }
1734 RocksReadSnapshotInner::Secondary(db) => {
1735 RocksDBRawIterEnum::ReadOnly((*db).raw_iterator_cf(cf))
1736 }
1737 }
1738 }
1739
1740 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
1742 let encoded_key = key.encode();
1743 let cf = self.cf_handle::<T>()?;
1744 let result = match &self.inner {
1745 RocksReadSnapshotInner::ReadWrite(snap) => snap.get_cf(cf, encoded_key.as_ref()),
1746 RocksReadSnapshotInner::Secondary(db) => db.get_cf(cf, encoded_key.as_ref()),
1747 }
1748 .map_err(|e| {
1749 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1750 message: e.to_string().into(),
1751 code: -1,
1752 }))
1753 })?;
1754
1755 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
1756 }
1757
1758 pub fn account_history_info(
1763 &self,
1764 address: Address,
1765 block_number: BlockNumber,
1766 lowest_available_block_number: Option<BlockNumber>,
1767 visible_tip: BlockNumber,
1768 ) -> ProviderResult<HistoryInfo> {
1769 let key = ShardedKey::new(address, block_number);
1770 self.history_info::<tables::AccountsHistory>(
1771 &self.accounts_history_iter,
1772 key.encode().as_ref(),
1773 block_number,
1774 lowest_available_block_number,
1775 visible_tip,
1776 |key_bytes| Ok(<ShardedKey<Address> as Decode>::decode(key_bytes)?.key == address),
1777 |prev_bytes| {
1778 <ShardedKey<Address> as Decode>::decode(prev_bytes)
1779 .map(|k| k.key == address)
1780 .unwrap_or(false)
1781 },
1782 )
1783 }
1784
1785 pub fn storage_history_info(
1790 &self,
1791 address: Address,
1792 storage_key: B256,
1793 block_number: BlockNumber,
1794 lowest_available_block_number: Option<BlockNumber>,
1795 visible_tip: BlockNumber,
1796 ) -> ProviderResult<HistoryInfo> {
1797 let key = StorageShardedKey::new(address, storage_key, block_number);
1798 self.history_info::<tables::StoragesHistory>(
1799 &self.storages_history_iter,
1800 key.encode().as_ref(),
1801 block_number,
1802 lowest_available_block_number,
1803 visible_tip,
1804 |key_bytes| {
1805 let k = <StorageShardedKey as Decode>::decode(key_bytes)?;
1806 Ok(k.address == address && k.sharded_key.key == storage_key)
1807 },
1808 |prev_bytes| {
1809 <StorageShardedKey as Decode>::decode(prev_bytes)
1810 .map(|k| k.address == address && k.sharded_key.key == storage_key)
1811 .unwrap_or(false)
1812 },
1813 )
1814 }
1815
1816 #[expect(clippy::too_many_arguments)]
1827 fn history_info<T>(
1828 &self,
1829 iter_cache: &Mutex<Option<RocksDBRawIterEnum<'db>>>,
1830 encoded_key: &[u8],
1831 block_number: BlockNumber,
1832 lowest_available_block_number: Option<BlockNumber>,
1833 visible_tip: BlockNumber,
1834 key_matches: impl FnOnce(&[u8]) -> Result<bool, reth_db_api::DatabaseError>,
1835 prev_key_matches: impl Fn(&[u8]) -> bool,
1836 ) -> ProviderResult<HistoryInfo>
1837 where
1838 T: Table<Value = BlockNumberList>,
1839 {
1840 let is_maybe_pruned = lowest_available_block_number.is_some();
1841 let fallback = || {
1842 Ok(if is_maybe_pruned {
1843 HistoryInfo::MaybeInPlainState
1844 } else {
1845 HistoryInfo::NotYetWritten
1846 })
1847 };
1848
1849 let cf = self.cf_handle::<T>()?;
1850 let mut guard = iter_cache.try_lock();
1853 let mut private_iter;
1854 let iter = match guard.as_mut() {
1855 Some(cached) => cached.get_or_insert_with(|| self.new_raw_iterator_cf(cf)),
1856 None => {
1857 private_iter = self.new_raw_iterator_cf(cf);
1858 &mut private_iter
1859 }
1860 };
1861
1862 iter.seek(encoded_key);
1863 iter.status().map_err(|e| {
1864 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1865 message: e.to_string().into(),
1866 code: -1,
1867 }))
1868 })?;
1869
1870 if !iter.valid() {
1871 return fallback();
1872 }
1873
1874 let Some(key_bytes) = iter.key() else {
1875 return fallback();
1876 };
1877 if !key_matches(key_bytes)? {
1878 return fallback();
1879 }
1880
1881 let Some(value_bytes) = iter.value() else {
1882 return fallback();
1883 };
1884 let chunk = BlockNumberList::decompress(value_bytes)?;
1885
1886 let (rank, found_block) = compute_history_rank(&chunk, block_number);
1887 let found_block = found_block.filter(|block| *block <= visible_tip);
1889
1890 let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) {
1891 iter.prev();
1892 iter.status().map_err(|e| {
1893 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1894 message: e.to_string().into(),
1895 code: -1,
1896 }))
1897 })?;
1898 let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches);
1899
1900 if found_block.is_none() && !has_prev {
1904 return fallback()
1905 }
1906
1907 !has_prev
1908 } else {
1909 false
1910 };
1911
1912 Ok(HistoryInfo::from_lookup(
1913 found_block,
1914 is_before_first_write,
1915 lowest_available_block_number,
1916 ))
1917 }
1918}
1919
1920pub struct OwnedRocksReadSnapshot {
1926 snapshot: Box<RocksReadSnapshot<'static>>,
1930 provider: RocksDBProvider,
1931}
1932
1933impl fmt::Debug for OwnedRocksReadSnapshot {
1934 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1935 f.debug_struct("OwnedRocksReadSnapshot").field("provider", &self.provider).finish()
1936 }
1937}
1938
1939impl OwnedRocksReadSnapshot {
1940 fn new(provider: &RocksDBProvider) -> Self {
1941 let provider = provider.clone();
1942 let snapshot = provider.snapshot();
1943 let snapshot = unsafe {
1948 std::mem::transmute::<RocksReadSnapshot<'_>, RocksReadSnapshot<'static>>(snapshot)
1949 };
1950 Self { snapshot: Box::new(snapshot), provider }
1951 }
1952
1953 pub fn as_snapshot(&self) -> &RocksReadSnapshot<'_> {
1955 unsafe {
1958 std::mem::transmute::<&RocksReadSnapshot<'static>, &RocksReadSnapshot<'_>>(
1959 &self.snapshot,
1960 )
1961 }
1962 }
1963}
1964
1965#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1967pub enum PruneShardOutcome {
1968 Deleted,
1970 Updated,
1972 Unchanged,
1974}
1975
1976#[derive(Debug, Default, Clone, Copy)]
1978pub struct PrunedIndices {
1979 pub deleted: usize,
1981 pub updated: usize,
1983 pub unchanged: usize,
1985}
1986
1987#[must_use = "batch must be committed"]
1997pub struct RocksDBBatch<'a> {
1998 provider: &'a RocksDBProvider,
1999 inner: WriteBatchWithTransaction<true>,
2000 buf: Vec<u8>,
2001 auto_commit_threshold: Option<usize>,
2003}
2004
2005impl fmt::Debug for RocksDBBatch<'_> {
2006 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2007 f.debug_struct("RocksDBBatch")
2008 .field("provider", &self.provider)
2009 .field("batch", &"<WriteBatchWithTransaction>")
2010 .field("length", &self.inner.len())
2012 .field("size_in_bytes", &self.inner.size_in_bytes())
2015 .finish()
2016 }
2017}
2018
2019impl<'a> RocksDBBatch<'a> {
2020 pub fn put<T: Table>(&mut self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
2024 let encoded_key = key.encode();
2025 self.put_encoded::<T>(&encoded_key, value)
2026 }
2027
2028 pub fn put_encoded<T: Table>(
2032 &mut self,
2033 key: &<T::Key as Encode>::Encoded,
2034 value: &T::Value,
2035 ) -> ProviderResult<()> {
2036 let value_bytes = compress_to_buf_or_ref!(self.buf, value).unwrap_or(&self.buf);
2037 self.inner.put_cf(self.provider.get_cf_handle::<T>()?, key, value_bytes);
2038 self.maybe_auto_commit()?;
2039 Ok(())
2040 }
2041
2042 pub fn delete<T: Table>(&mut self, key: T::Key) -> ProviderResult<()> {
2046 self.inner.delete_cf(self.provider.get_cf_handle::<T>()?, key.encode().as_ref());
2047 self.maybe_auto_commit()?;
2048 Ok(())
2049 }
2050
2051 fn maybe_auto_commit(&mut self) -> ProviderResult<()> {
2056 if let Some(threshold) = self.auto_commit_threshold &&
2057 self.inner.size_in_bytes() >= threshold
2058 {
2059 tracing::debug!(
2060 target: "providers::rocksdb",
2061 batch_size = self.inner.size_in_bytes(),
2062 threshold,
2063 "Auto-committing RocksDB batch"
2064 );
2065 let old_batch = std::mem::take(&mut self.inner);
2066 self.provider.0.db_rw().write_opt(old_batch, &synced_write_options()).map_err(|e| {
2067 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2068 message: e.to_string().into(),
2069 code: -1,
2070 }))
2071 })?;
2072 }
2073 Ok(())
2074 }
2075
2076 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = self.inner.len(), batch_size = self.inner.size_in_bytes()))]
2083 pub fn commit(self) -> ProviderResult<()> {
2084 self.provider.0.db_rw().write_opt(self.inner, &synced_write_options()).map_err(|e| {
2085 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2086 message: e.to_string().into(),
2087 code: -1,
2088 }))
2089 })
2090 }
2091
2092 pub fn len(&self) -> usize {
2094 self.inner.len()
2095 }
2096
2097 pub fn is_empty(&self) -> bool {
2099 self.inner.is_empty()
2100 }
2101
2102 pub fn size_in_bytes(&self) -> usize {
2104 self.inner.size_in_bytes()
2105 }
2106
2107 pub const fn provider(&self) -> &RocksDBProvider {
2109 self.provider
2110 }
2111
2112 pub fn into_inner(self) -> WriteBatchWithTransaction<true> {
2116 self.inner
2117 }
2118
2119 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
2124 self.provider.get::<T>(key)
2125 }
2126
2127 pub fn append_account_history_shard(
2139 &mut self,
2140 address: Address,
2141 indices: impl IntoIterator<Item = u64>,
2142 ) -> ProviderResult<()> {
2143 let indices: Vec<u64> = indices.into_iter().collect();
2144
2145 if indices.is_empty() {
2146 return Ok(());
2147 }
2148
2149 debug_assert!(
2150 indices.windows(2).all(|w| w[0] < w[1]),
2151 "indices must be strictly increasing: {:?}",
2152 indices
2153 );
2154
2155 let last_key = ShardedKey::new(address, u64::MAX);
2156 let last_shard_opt = self.provider.get::<tables::AccountsHistory>(last_key.clone())?;
2157 let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
2158
2159 last_shard.append(indices).map_err(ProviderError::other)?;
2160
2161 if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
2163 self.put::<tables::AccountsHistory>(last_key, &last_shard)?;
2164 return Ok(());
2165 }
2166
2167 let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
2169 let mut chunks_peekable = chunks.into_iter().peekable();
2170
2171 while let Some(chunk) = chunks_peekable.next() {
2172 let shard = BlockNumberList::new_pre_sorted(chunk);
2173 let highest_block_number = if chunks_peekable.peek().is_some() {
2174 shard.iter().next_back().expect("`chunks` does not return empty list")
2175 } else {
2176 u64::MAX
2177 };
2178
2179 self.put::<tables::AccountsHistory>(
2180 ShardedKey::new(address, highest_block_number),
2181 &shard,
2182 )?;
2183 }
2184
2185 Ok(())
2186 }
2187
2188 pub fn append_storage_history_shard(
2200 &mut self,
2201 address: Address,
2202 storage_key: B256,
2203 indices: impl IntoIterator<Item = u64>,
2204 ) -> ProviderResult<()> {
2205 let indices: Vec<u64> = indices.into_iter().collect();
2206
2207 for (key, shard) in
2208 self.provider.storage_history_shards_to_put(address, storage_key, indices)?
2209 {
2210 self.put::<tables::StoragesHistory>(key, &shard)?;
2211 }
2212
2213 Ok(())
2214 }
2215
2216 pub fn unwind_account_history_to(
2223 &mut self,
2224 address: Address,
2225 keep_to: BlockNumber,
2226 ) -> ProviderResult<()> {
2227 let shards = self.provider.account_history_shards(address)?;
2228 if shards.is_empty() {
2229 return Ok(());
2230 }
2231
2232 let boundary_idx = shards.iter().position(|(key, _)| {
2235 key.highest_block_number == u64::MAX || key.highest_block_number > keep_to
2236 });
2237
2238 let Some(boundary_idx) = boundary_idx else {
2240 let (last_key, last_value) = shards.last().expect("shards is non-empty");
2241 if last_key.highest_block_number != u64::MAX {
2242 self.delete::<tables::AccountsHistory>(last_key.clone())?;
2243 self.put::<tables::AccountsHistory>(
2244 ShardedKey::new(address, u64::MAX),
2245 last_value,
2246 )?;
2247 }
2248 return Ok(());
2249 };
2250
2251 for (key, _) in shards.iter().skip(boundary_idx + 1) {
2253 self.delete::<tables::AccountsHistory>(key.clone())?;
2254 }
2255
2256 let (boundary_key, boundary_list) = &shards[boundary_idx];
2258
2259 self.delete::<tables::AccountsHistory>(boundary_key.clone())?;
2261
2262 let new_last =
2264 BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2265
2266 if new_last.is_empty() {
2267 if boundary_idx == 0 {
2270 return Ok(());
2272 }
2273
2274 let (prev_key, prev_value) = &shards[boundary_idx - 1];
2275 if prev_key.highest_block_number != u64::MAX {
2276 self.delete::<tables::AccountsHistory>(prev_key.clone())?;
2277 self.put::<tables::AccountsHistory>(
2278 ShardedKey::new(address, u64::MAX),
2279 prev_value,
2280 )?;
2281 }
2282 return Ok(());
2283 }
2284
2285 self.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &new_last)?;
2286
2287 Ok(())
2288 }
2289
2290 #[expect(clippy::too_many_arguments)]
2299 fn prune_history_shards_inner<K>(
2300 &mut self,
2301 shards: Vec<(K, BlockNumberList)>,
2302 shards_complete: bool,
2303 to_block: BlockNumber,
2304 get_highest: impl Fn(&K) -> u64,
2305 is_sentinel: impl Fn(&K) -> bool,
2306 delete_shard: impl Fn(&mut Self, K) -> ProviderResult<()>,
2307 put_shard: impl Fn(&mut Self, K, &BlockNumberList) -> ProviderResult<()>,
2308 create_sentinel: impl Fn() -> K,
2309 ) -> ProviderResult<PruneShardOutcome>
2310 where
2311 K: Clone,
2312 {
2313 if shards.is_empty() {
2314 return Ok(PruneShardOutcome::Unchanged);
2315 }
2316
2317 let mut deleted = false;
2318 let mut updated = false;
2319 let mut last_remaining: Option<(K, BlockNumberList)> = None;
2320
2321 for (key, mut block_list) in shards {
2322 if !is_sentinel(&key) && get_highest(&key) <= to_block {
2323 delete_shard(self, key)?;
2324 deleted = true;
2325 } else {
2326 let removed = block_list.remove_range(0..=to_block);
2327
2328 if block_list.is_empty() {
2329 delete_shard(self, key)?;
2330 deleted = true;
2331 } else if removed > 0 {
2332 put_shard(self, key.clone(), &block_list)?;
2333 last_remaining = Some((key, block_list));
2334 updated = true;
2335 } else {
2336 last_remaining = Some((key, block_list));
2337 }
2338 }
2339 }
2340
2341 if shards_complete &&
2342 let Some((last_key, last_value)) = last_remaining &&
2343 !is_sentinel(&last_key)
2344 {
2345 delete_shard(self, last_key)?;
2346 put_shard(self, create_sentinel(), &last_value)?;
2347 updated = true;
2348 }
2349
2350 if deleted {
2351 Ok(PruneShardOutcome::Deleted)
2352 } else if updated {
2353 Ok(PruneShardOutcome::Updated)
2354 } else {
2355 Ok(PruneShardOutcome::Unchanged)
2356 }
2357 }
2358
2359 pub fn prune_account_history_to(
2364 &mut self,
2365 address: Address,
2366 to_block: BlockNumber,
2367 ) -> ProviderResult<PruneShardOutcome> {
2368 let shards = self.provider.account_history_shards(address)?;
2369 self.prune_history_shards_inner(
2370 shards,
2371 true,
2372 to_block,
2373 |key| key.highest_block_number,
2374 |key| key.highest_block_number == u64::MAX,
2375 |batch, key| batch.delete::<tables::AccountsHistory>(key),
2376 |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2377 || ShardedKey::new(address, u64::MAX),
2378 )
2379 }
2380
2381 pub fn prune_account_history_batch(
2390 &mut self,
2391 targets: &[(Address, BlockNumber)],
2392 ) -> ProviderResult<PrunedIndices> {
2393 if targets.is_empty() {
2394 return Ok(PrunedIndices::default());
2395 }
2396
2397 debug_assert!(
2398 targets.windows(2).all(|w| w[0].0 < w[1].0),
2399 "prune_account_history_batch: targets must be sorted and unique"
2400 );
2401
2402 const PREFIX_LEN: usize = 20;
2405
2406 let cf = self.provider.get_cf_handle::<tables::AccountsHistory>()?;
2407 let mut iter = self.provider.0.raw_iterator_cf(cf);
2408 let mut outcomes = PrunedIndices::default();
2409
2410 for (address, to_block) in targets {
2411 let start_key = ShardedKey::new(*address, 0u64).encode();
2413 let target_prefix = &start_key[..PREFIX_LEN];
2414
2415 let needs_seek = if iter.valid() {
2423 iter.key().is_none_or(|current_key| {
2427 current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2428 })
2429 } else {
2430 true
2431 };
2432
2433 if needs_seek {
2434 iter.seek(start_key);
2435 iter.status().map_err(|e| {
2436 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2437 message: e.to_string().into(),
2438 code: -1,
2439 }))
2440 })?;
2441 }
2442
2443 let mut shards = Vec::new();
2446 let mut shards_complete = true;
2447 while iter.valid() {
2448 let Some(key_bytes) = iter.key() else { break };
2449
2450 let current_prefix = key_bytes.get(..PREFIX_LEN);
2452 if current_prefix != Some(target_prefix) {
2453 break;
2454 }
2455
2456 let key = ShardedKey::<Address>::decode(key_bytes)
2458 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2459
2460 let Some(value_bytes) = iter.value() else { break };
2461 let value = BlockNumberList::decompress(value_bytes)
2462 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2463
2464 let highest = key.highest_block_number;
2465 shards.push((key, value));
2466
2467 iter.next();
2468
2469 if highest > *to_block {
2474 shards_complete = iter.key().and_then(|next_key| next_key.get(..PREFIX_LEN)) !=
2475 Some(target_prefix);
2476 break;
2477 }
2478 }
2479
2480 if !iter.valid() {
2483 iter.status().map_err(|e| {
2484 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2485 message: e.to_string().into(),
2486 code: -1,
2487 }))
2488 })?;
2489 }
2490
2491 match self.prune_history_shards_inner(
2492 shards,
2493 shards_complete,
2494 *to_block,
2495 |key| key.highest_block_number,
2496 |key| key.highest_block_number == u64::MAX,
2497 |batch, key| batch.delete::<tables::AccountsHistory>(key),
2498 |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2499 || ShardedKey::new(*address, u64::MAX),
2500 )? {
2501 PruneShardOutcome::Deleted => outcomes.deleted += 1,
2502 PruneShardOutcome::Updated => outcomes.updated += 1,
2503 PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2504 }
2505 }
2506
2507 Ok(outcomes)
2508 }
2509
2510 pub fn prune_storage_history_to(
2516 &mut self,
2517 address: Address,
2518 storage_key: B256,
2519 to_block: BlockNumber,
2520 ) -> ProviderResult<PruneShardOutcome> {
2521 let shards = self.provider.storage_history_shards(address, storage_key)?;
2522 self.prune_history_shards_inner(
2523 shards,
2524 true,
2525 to_block,
2526 |key| key.sharded_key.highest_block_number,
2527 |key| key.sharded_key.highest_block_number == u64::MAX,
2528 |batch, key| batch.delete::<tables::StoragesHistory>(key),
2529 |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2530 || StorageShardedKey::last(address, storage_key),
2531 )
2532 }
2533
2534 pub fn prune_storage_history_batch(
2544 &mut self,
2545 targets: &[((Address, B256), BlockNumber)],
2546 ) -> ProviderResult<PrunedIndices> {
2547 if targets.is_empty() {
2548 return Ok(PrunedIndices::default());
2549 }
2550
2551 debug_assert!(
2552 targets.windows(2).all(|w| w[0].0 < w[1].0),
2553 "prune_storage_history_batch: targets must be sorted and unique"
2554 );
2555
2556 const PREFIX_LEN: usize = 52;
2559
2560 let cf = self.provider.get_cf_handle::<tables::StoragesHistory>()?;
2561 let mut iter = self.provider.0.raw_iterator_cf(cf);
2562 let mut outcomes = PrunedIndices::default();
2563
2564 for ((address, storage_key), to_block) in targets {
2565 let start_key = StorageShardedKey::new(*address, *storage_key, 0u64).encode();
2567 let target_prefix = &start_key[..PREFIX_LEN];
2568
2569 let needs_seek = if iter.valid() {
2577 iter.key().is_none_or(|current_key| {
2581 current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2582 })
2583 } else {
2584 true
2585 };
2586
2587 if needs_seek {
2588 iter.seek(start_key);
2589 iter.status().map_err(|e| {
2590 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2591 message: e.to_string().into(),
2592 code: -1,
2593 }))
2594 })?;
2595 }
2596
2597 let mut shards = Vec::new();
2600 let mut shards_complete = true;
2601 while iter.valid() {
2602 let Some(key_bytes) = iter.key() else { break };
2603
2604 let current_prefix = key_bytes.get(..PREFIX_LEN);
2606 if current_prefix != Some(target_prefix) {
2607 break;
2608 }
2609
2610 let key = StorageShardedKey::decode(key_bytes)
2612 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2613
2614 let Some(value_bytes) = iter.value() else { break };
2615 let value = BlockNumberList::decompress(value_bytes)
2616 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2617
2618 let highest = key.sharded_key.highest_block_number;
2619 shards.push((key, value));
2620
2621 iter.next();
2622
2623 if highest > *to_block {
2628 shards_complete = iter.key().and_then(|next_key| next_key.get(..PREFIX_LEN)) !=
2629 Some(target_prefix);
2630 break;
2631 }
2632 }
2633
2634 if !iter.valid() {
2637 iter.status().map_err(|e| {
2638 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2639 message: e.to_string().into(),
2640 code: -1,
2641 }))
2642 })?;
2643 }
2644
2645 match self.prune_history_shards_inner(
2647 shards,
2648 shards_complete,
2649 *to_block,
2650 |key| key.sharded_key.highest_block_number,
2651 |key| key.sharded_key.highest_block_number == u64::MAX,
2652 |batch, key| batch.delete::<tables::StoragesHistory>(key),
2653 |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2654 || StorageShardedKey::last(*address, *storage_key),
2655 )? {
2656 PruneShardOutcome::Deleted => outcomes.deleted += 1,
2657 PruneShardOutcome::Updated => outcomes.updated += 1,
2658 PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2659 }
2660 }
2661
2662 Ok(outcomes)
2663 }
2664
2665 pub fn unwind_storage_history_to(
2674 &mut self,
2675 address: Address,
2676 storage_key: B256,
2677 keep_to: BlockNumber,
2678 ) -> ProviderResult<()> {
2679 let shards = self.provider.storage_history_shards(address, storage_key)?;
2680 if shards.is_empty() {
2681 return Ok(());
2682 }
2683
2684 let boundary_idx = shards.iter().position(|(key, _)| {
2687 key.sharded_key.highest_block_number == u64::MAX ||
2688 key.sharded_key.highest_block_number > keep_to
2689 });
2690
2691 let Some(boundary_idx) = boundary_idx else {
2693 let (last_key, last_value) = shards.last().expect("shards is non-empty");
2694 if last_key.sharded_key.highest_block_number != u64::MAX {
2695 self.delete::<tables::StoragesHistory>(last_key.clone())?;
2696 self.put::<tables::StoragesHistory>(
2697 StorageShardedKey::last(address, storage_key),
2698 last_value,
2699 )?;
2700 }
2701 return Ok(());
2702 };
2703
2704 for (key, _) in shards.iter().skip(boundary_idx + 1) {
2706 self.delete::<tables::StoragesHistory>(key.clone())?;
2707 }
2708
2709 let (boundary_key, boundary_list) = &shards[boundary_idx];
2711
2712 self.delete::<tables::StoragesHistory>(boundary_key.clone())?;
2714
2715 let new_last =
2717 BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2718
2719 if new_last.is_empty() {
2720 if boundary_idx == 0 {
2723 return Ok(());
2725 }
2726
2727 let (prev_key, prev_value) = &shards[boundary_idx - 1];
2728 if prev_key.sharded_key.highest_block_number != u64::MAX {
2729 self.delete::<tables::StoragesHistory>(prev_key.clone())?;
2730 self.put::<tables::StoragesHistory>(
2731 StorageShardedKey::last(address, storage_key),
2732 prev_value,
2733 )?;
2734 }
2735 return Ok(());
2736 }
2737
2738 self.put::<tables::StoragesHistory>(
2739 StorageShardedKey::last(address, storage_key),
2740 &new_last,
2741 )?;
2742
2743 Ok(())
2744 }
2745
2746 pub fn clear_account_history(&mut self, address: Address) -> ProviderResult<()> {
2750 let shards = self.provider.account_history_shards(address)?;
2751 for (key, _) in shards {
2752 self.delete::<tables::AccountsHistory>(key)?;
2753 }
2754 Ok(())
2755 }
2756
2757 pub fn clear_storage_history(
2761 &mut self,
2762 address: Address,
2763 storage_key: B256,
2764 ) -> ProviderResult<()> {
2765 let shards = self.provider.storage_history_shards(address, storage_key)?;
2766 for (key, _) in shards {
2767 self.delete::<tables::StoragesHistory>(key)?;
2768 }
2769 Ok(())
2770 }
2771}
2772
2773pub struct RocksTx<'db> {
2783 inner: Transaction<'db, OptimisticTransactionDB>,
2784 provider: &'db RocksDBProvider,
2785}
2786
2787impl fmt::Debug for RocksTx<'_> {
2788 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2789 f.debug_struct("RocksTx").field("provider", &self.provider).finish_non_exhaustive()
2790 }
2791}
2792
2793impl<'db> RocksTx<'db> {
2794 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
2796 let encoded_key = key.encode();
2797 self.get_encoded::<T>(&encoded_key)
2798 }
2799
2800 pub fn get_encoded<T: Table>(
2802 &self,
2803 key: &<T::Key as Encode>::Encoded,
2804 ) -> ProviderResult<Option<T::Value>> {
2805 let cf = self.provider.get_cf_handle::<T>()?;
2806 let result = self.inner.get_cf(cf, key.as_ref()).map_err(|e| {
2807 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2808 message: e.to_string().into(),
2809 code: -1,
2810 }))
2811 })?;
2812
2813 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
2814 }
2815
2816 pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
2818 let encoded_key = key.encode();
2819 self.put_encoded::<T>(&encoded_key, value)
2820 }
2821
2822 pub fn put_encoded<T: Table>(
2824 &self,
2825 key: &<T::Key as Encode>::Encoded,
2826 value: &T::Value,
2827 ) -> ProviderResult<()> {
2828 let cf = self.provider.get_cf_handle::<T>()?;
2829 let mut buf = Vec::new();
2830 let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
2831
2832 self.inner.put_cf(cf, key.as_ref(), value_bytes).map_err(|e| {
2833 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
2834 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
2835 operation: DatabaseWriteOperation::PutUpsert,
2836 table_name: T::NAME,
2837 key: key.as_ref().to_vec(),
2838 })))
2839 })
2840 }
2841
2842 pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
2844 let cf = self.provider.get_cf_handle::<T>()?;
2845 self.inner.delete_cf(cf, key.encode().as_ref()).map_err(|e| {
2846 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
2847 message: e.to_string().into(),
2848 code: -1,
2849 }))
2850 })
2851 }
2852
2853 pub fn iter<T: Table>(&self) -> ProviderResult<RocksTxIter<'_, T>> {
2857 let cf = self.provider.get_cf_handle::<T>()?;
2858 let iter = self.inner.iterator_cf(cf, IteratorMode::Start);
2859 Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2860 }
2861
2862 pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksTxIter<'_, T>> {
2864 let cf = self.provider.get_cf_handle::<T>()?;
2865 let encoded_key = key.encode();
2866 let iter = self
2867 .inner
2868 .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
2869 Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2870 }
2871
2872 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2874 pub fn commit(self) -> ProviderResult<()> {
2875 self.inner.commit().map_err(|e| {
2876 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2877 message: e.to_string().into(),
2878 code: -1,
2879 }))
2880 })
2881 }
2882
2883 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2885 pub fn rollback(self) -> ProviderResult<()> {
2886 self.inner.rollback().map_err(|e| {
2887 ProviderError::Database(DatabaseError::Other(format!("rollback failed: {e}")))
2888 })
2889 }
2890}
2891
2892enum RocksDBIterEnum<'db> {
2894 ReadWrite(rocksdb::DBIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2896 ReadOnly(rocksdb::DBIteratorWithThreadMode<'db, DB>),
2898}
2899
2900impl Iterator for RocksDBIterEnum<'_> {
2901 type Item = Result<(Box<[u8]>, Box<[u8]>), rocksdb::Error>;
2902
2903 fn next(&mut self) -> Option<Self::Item> {
2904 match self {
2905 Self::ReadWrite(iter) => iter.next(),
2906 Self::ReadOnly(iter) => iter.next(),
2907 }
2908 }
2909}
2910
2911enum RocksDBRawIterEnum<'db> {
2916 ReadWrite(DBRawIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2918 ReadOnly(DBRawIteratorWithThreadMode<'db, DB>),
2920}
2921
2922impl RocksDBRawIterEnum<'_> {
2923 fn seek(&mut self, key: impl AsRef<[u8]>) {
2925 match self {
2926 Self::ReadWrite(iter) => iter.seek(key),
2927 Self::ReadOnly(iter) => iter.seek(key),
2928 }
2929 }
2930
2931 fn valid(&self) -> bool {
2933 match self {
2934 Self::ReadWrite(iter) => iter.valid(),
2935 Self::ReadOnly(iter) => iter.valid(),
2936 }
2937 }
2938
2939 fn key(&self) -> Option<&[u8]> {
2941 match self {
2942 Self::ReadWrite(iter) => iter.key(),
2943 Self::ReadOnly(iter) => iter.key(),
2944 }
2945 }
2946
2947 fn value(&self) -> Option<&[u8]> {
2949 match self {
2950 Self::ReadWrite(iter) => iter.value(),
2951 Self::ReadOnly(iter) => iter.value(),
2952 }
2953 }
2954
2955 fn next(&mut self) {
2957 match self {
2958 Self::ReadWrite(iter) => iter.next(),
2959 Self::ReadOnly(iter) => iter.next(),
2960 }
2961 }
2962
2963 fn prev(&mut self) {
2965 match self {
2966 Self::ReadWrite(iter) => iter.prev(),
2967 Self::ReadOnly(iter) => iter.prev(),
2968 }
2969 }
2970
2971 fn status(&self) -> Result<(), rocksdb::Error> {
2973 match self {
2974 Self::ReadWrite(iter) => iter.status(),
2975 Self::ReadOnly(iter) => iter.status(),
2976 }
2977 }
2978}
2979
2980pub struct RocksDBIter<'db, T: Table> {
2984 inner: RocksDBIterEnum<'db>,
2985 _marker: std::marker::PhantomData<T>,
2986}
2987
2988impl<T: Table> fmt::Debug for RocksDBIter<'_, T> {
2989 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2990 f.debug_struct("RocksDBIter").field("table", &T::NAME).finish_non_exhaustive()
2991 }
2992}
2993
2994impl<T: Table> Iterator for RocksDBIter<'_, T> {
2995 type Item = ProviderResult<(T::Key, T::Value)>;
2996
2997 fn next(&mut self) -> Option<Self::Item> {
2998 Some(decode_iter_item::<T>(self.inner.next()?))
2999 }
3000}
3001
3002pub struct RocksDBRawIter<'db> {
3006 inner: RocksDBIterEnum<'db>,
3007}
3008
3009impl fmt::Debug for RocksDBRawIter<'_> {
3010 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3011 f.debug_struct("RocksDBRawIter").finish_non_exhaustive()
3012 }
3013}
3014
3015impl Iterator for RocksDBRawIter<'_> {
3016 type Item = ProviderResult<(Box<[u8]>, Box<[u8]>)>;
3017
3018 fn next(&mut self) -> Option<Self::Item> {
3019 match self.inner.next()? {
3020 Ok(kv) => Some(Ok(kv)),
3021 Err(e) => Some(Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
3022 message: e.to_string().into(),
3023 code: -1,
3024 })))),
3025 }
3026 }
3027}
3028
3029pub(crate) struct RocksDBRawKeyIter<'db> {
3031 inner: RocksDBRawIterEnum<'db>,
3032}
3033
3034impl fmt::Debug for RocksDBRawKeyIter<'_> {
3035 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3036 f.debug_struct("RocksDBRawKeyIter").finish_non_exhaustive()
3037 }
3038}
3039
3040impl Iterator for RocksDBRawKeyIter<'_> {
3041 type Item = ProviderResult<Box<[u8]>>;
3042
3043 fn next(&mut self) -> Option<Self::Item> {
3044 if !self.inner.valid() {
3045 return self.inner.status().err().map(|e| {
3046 Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
3047 message: e.to_string().into(),
3048 code: -1,
3049 })))
3050 })
3051 }
3052
3053 let Some(key) = self.inner.key() else {
3054 return Some(Err(ProviderError::Database(DatabaseError::Decode)))
3055 };
3056 let key = Box::from(key);
3057 self.inner.next();
3058 Some(Ok(key))
3059 }
3060}
3061
3062pub struct RocksTxIter<'tx, T: Table> {
3066 inner: rocksdb::DBIteratorWithThreadMode<'tx, Transaction<'tx, OptimisticTransactionDB>>,
3067 _marker: std::marker::PhantomData<T>,
3068}
3069
3070impl<T: Table> fmt::Debug for RocksTxIter<'_, T> {
3071 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3072 f.debug_struct("RocksTxIter").field("table", &T::NAME).finish_non_exhaustive()
3073 }
3074}
3075
3076impl<T: Table> Iterator for RocksTxIter<'_, T> {
3077 type Item = ProviderResult<(T::Key, T::Value)>;
3078
3079 fn next(&mut self) -> Option<Self::Item> {
3080 Some(decode_iter_item::<T>(self.inner.next()?))
3081 }
3082}
3083
3084fn decode_iter_item<T: Table>(result: RawKVResult) -> ProviderResult<(T::Key, T::Value)> {
3089 let (key_bytes, value_bytes) = result.map_err(|e| {
3090 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
3091 message: e.to_string().into(),
3092 code: -1,
3093 }))
3094 })?;
3095
3096 let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
3097 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
3098
3099 let value = T::Value::decompress(&value_bytes)
3100 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
3101
3102 Ok((key, value))
3103}
3104
3105const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
3107 match level {
3108 LogLevel::Fatal => rocksdb::LogLevel::Fatal,
3109 LogLevel::Error => rocksdb::LogLevel::Error,
3110 LogLevel::Warn => rocksdb::LogLevel::Warn,
3111 LogLevel::Notice | LogLevel::Verbose => rocksdb::LogLevel::Info,
3112 LogLevel::Debug | LogLevel::Trace | LogLevel::Extra => rocksdb::LogLevel::Debug,
3113 }
3114}
3115
3116fn select_max_open_files() -> i32 {
3133 let file_descriptor_limit = current_file_descriptor_limit();
3134 let max_open_files = max_open_files_for_limit(file_descriptor_limit);
3135
3136 if max_open_files == LIMITED_MAX_OPEN_FILES {
3137 tracing::warn!(
3138 target: "providers::rocksdb",
3139 ?file_descriptor_limit,
3140 threshold = HIGH_FILE_DESCRIPTOR_LIMIT,
3141 max_open_files,
3142 "RocksDB will not keep all files open; performance may be reduced"
3143 );
3144 }
3145
3146 max_open_files
3147}
3148
3149const fn max_open_files_for_limit(file_descriptor_limit: Option<u64>) -> i32 {
3150 match file_descriptor_limit {
3151 Some(limit) if limit >= HIGH_FILE_DESCRIPTOR_LIMIT => KEEP_ALL_FILES_OPEN,
3152 _ => LIMITED_MAX_OPEN_FILES,
3153 }
3154}
3155
3156#[cfg(unix)]
3157#[allow(clippy::useless_conversion)]
3158fn current_file_descriptor_limit() -> Option<u64> {
3159 let mut limit = libc::rlimit { rlim_cur: 0, rlim_max: 0 };
3160 let result = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) };
3162 (result == 0).then(|| limit.rlim_cur.into())
3163}
3164
3165#[cfg(not(unix))]
3166const fn current_file_descriptor_limit() -> Option<u64> {
3167 None
3168}
3169
3170#[cfg(test)]
3171mod tests {
3172 use super::*;
3173 use crate::providers::HistoryInfo;
3174 use alloy_primitives::{Address, Bytes, TxHash, B256};
3175 use reth_db_api::{
3176 models::{
3177 sharded_key::{ShardedKey, NUM_OF_INDICES_IN_SHARD},
3178 storage_sharded_key::StorageShardedKey,
3179 IntegerList,
3180 },
3181 table::Table,
3182 tables,
3183 };
3184 use rocksdb::{
3185 perf::{set_perf_stats, PerfContext, PerfMetric, PerfStatsLevel},
3186 statistics::{Histogram, Ticker},
3187 };
3188 use tempfile::TempDir;
3189
3190 fn deleted_keys_skipped<R>(f: impl FnOnce() -> R) -> (R, u64) {
3192 set_perf_stats(PerfStatsLevel::EnableCount);
3193 let mut ctx = PerfContext::default();
3194 ctx.reset();
3195 let result = f();
3196 let skipped = ctx.metric(PerfMetric::InternalDeleteSkippedCount);
3197 set_perf_stats(PerfStatsLevel::Disable);
3198 (result, skipped)
3199 }
3200
3201 fn address_at(i: u64) -> Address {
3202 let mut bytes = [0u8; 20];
3203 bytes[12..].copy_from_slice(&i.to_be_bytes());
3204 Address::from(bytes)
3205 }
3206
3207 #[test]
3212 fn test_history_shards_lookup_skips_only_own_deleted_keys() {
3213 const ADDRESSES: u64 = 5_000;
3214
3215 let temp_dir = TempDir::new().unwrap();
3216 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3217 let storage_key = B256::with_last_byte(1);
3218 let list = BlockNumberList::new_pre_sorted([1]);
3219
3220 let mut batch = provider.batch();
3222 for i in 0..ADDRESSES {
3223 let address = address_at(i);
3224 batch
3225 .put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &list)
3226 .unwrap();
3227 batch
3228 .put::<tables::StoragesHistory>(
3229 StorageShardedKey::new(address, storage_key, u64::MAX),
3230 &list,
3231 )
3232 .unwrap();
3233 }
3234 let live = Address::repeat_byte(0xff);
3235 batch.put::<tables::AccountsHistory>(ShardedKey::new(live, u64::MAX), &list).unwrap();
3236 batch
3237 .put::<tables::StoragesHistory>(
3238 StorageShardedKey::new(live, storage_key, u64::MAX),
3239 &list,
3240 )
3241 .unwrap();
3242 batch.commit().unwrap();
3243
3244 let mut batch = provider.batch();
3246 for i in 0..ADDRESSES {
3247 batch.clear_account_history(address_at(i)).unwrap();
3248 batch.clear_storage_history(address_at(i), storage_key).unwrap();
3249 }
3250 batch.commit().unwrap();
3251
3252 let (shards, skipped) =
3253 deleted_keys_skipped(|| provider.account_history_shards(address_at(0)));
3254 assert!(shards.unwrap().is_empty());
3255 assert!(skipped <= 1, "account lookup skipped {skipped} deleted keys");
3256
3257 let (shards, skipped) =
3258 deleted_keys_skipped(|| provider.storage_history_shards(address_at(0), storage_key));
3259 assert!(shards.unwrap().is_empty());
3260 assert!(skipped <= 1, "storage lookup skipped {skipped} deleted keys");
3261
3262 assert_eq!(provider.account_history_shards(live).unwrap().len(), 1);
3264 assert_eq!(provider.storage_history_shards(live, storage_key).unwrap().len(), 1);
3265 }
3266
3267 #[test]
3268 fn max_open_files_adapts_to_file_descriptor_limit() {
3269 assert_eq!(max_open_files_for_limit(Some(HIGH_FILE_DESCRIPTOR_LIMIT)), KEEP_ALL_FILES_OPEN);
3270 assert_eq!(
3271 max_open_files_for_limit(Some(HIGH_FILE_DESCRIPTOR_LIMIT - 1)),
3272 LIMITED_MAX_OPEN_FILES
3273 );
3274 assert_eq!(max_open_files_for_limit(None), LIMITED_MAX_OPEN_FILES);
3275 }
3276
3277 #[test]
3278 fn test_with_default_tables_registers_required_column_families() {
3279 let temp_dir = TempDir::new().unwrap();
3280
3281 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3283
3284 let tx_hash = TxHash::from(B256::repeat_byte(1u8));
3286 provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3287 assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3288
3289 let key = ShardedKey::new(Address::ZERO, 100);
3291 let value = IntegerList::default();
3292 provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
3293 assert!(provider.get::<tables::AccountsHistory>(key).unwrap().is_some());
3294
3295 let key = StorageShardedKey::new(Address::ZERO, B256::ZERO, 100);
3297 provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
3298 assert!(provider.get::<tables::StoragesHistory>(key).unwrap().is_some());
3299
3300 let bal_key =
3301 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3302 let bal_value =
3303 reth_db_api::models::StoredBlockAccessList::new(Bytes::from_static(&[0xc0]));
3304 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3305 assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
3306 provider
3307 .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
3308 .unwrap();
3309 assert_eq!(
3310 provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
3311 Some(bal_key.number())
3312 );
3313 }
3314
3315 #[test]
3316 fn block_access_lists_store_large_payloads_in_blob_files() {
3317 let temp_dir = TempDir::new().unwrap();
3318 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3319 let bal_key =
3320 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3321 let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
3322 0;
3323 DEFAULT_BAL_MIN_BLOB_SIZE as usize +
3324 1
3325 ]));
3326
3327 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3328 provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3329
3330 let has_blob_file = std::fs::read_dir(temp_dir.path()).unwrap().any(|entry| {
3331 entry.unwrap().path().extension().is_some_and(|extension| extension == "blob")
3332 });
3333 assert!(has_blob_file);
3334 }
3335
3336 #[derive(Debug)]
3337 struct TestTable;
3338
3339 impl Table for TestTable {
3340 const NAME: &'static str = "TestTable";
3341 const DUPSORT: bool = false;
3342 type Key = u64;
3343 type Value = Vec<u8>;
3344 }
3345
3346 #[test]
3347 fn test_reopens_with_unknown_column_family() {
3348 let temp_dir = TempDir::new().unwrap();
3349 let value = b"test_value".to_vec();
3350
3351 let provider = RocksDBBuilder::new(temp_dir.path())
3352 .with_default_tables()
3353 .with_table::<TestTable>()
3354 .build()
3355 .unwrap();
3356 provider.put::<TestTable>(42, &value).unwrap();
3357 drop(provider);
3358
3359 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3360 assert_eq!(provider.get::<TestTable>(42).unwrap(), Some(value));
3361 }
3362
3363 #[test]
3364 fn test_reopens_blob_column_family_with_legacy_table_set() {
3365 let temp_dir = TempDir::new().unwrap();
3366 let bal_key =
3367 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3368 let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
3369 0;
3370 DEFAULT_BAL_MIN_BLOB_SIZE as usize +
3371 1
3372 ]));
3373
3374 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3375 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3376 provider
3377 .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
3378 .unwrap();
3379 provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3380 drop(provider);
3381
3382 let provider = RocksDBBuilder::new(temp_dir.path())
3383 .with_table::<tables::TransactionHashNumbers>()
3384 .with_table::<tables::AccountsHistory>()
3385 .with_table::<tables::StoragesHistory>()
3386 .build()
3387 .unwrap();
3388 assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
3389 assert_eq!(
3390 provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
3391 Some(bal_key.number())
3392 );
3393 }
3394
3395 #[test]
3396 fn test_basic_operations() {
3397 let temp_dir = TempDir::new().unwrap();
3398
3399 let provider = RocksDBBuilder::new(temp_dir.path())
3400 .with_table::<TestTable>() .build()
3402 .unwrap();
3403
3404 let key = 42u64;
3405 let value = b"test_value".to_vec();
3406
3407 provider.put::<TestTable>(key, &value).unwrap();
3409
3410 let result = provider.get::<TestTable>(key).unwrap();
3412 assert_eq!(result, Some(value));
3413
3414 provider.delete::<TestTable>(key).unwrap();
3416
3417 assert_eq!(provider.get::<TestTable>(key).unwrap(), None);
3419 }
3420
3421 #[test]
3422 fn test_batch_operations() {
3423 let temp_dir = TempDir::new().unwrap();
3424 let provider =
3425 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3426
3427 provider
3429 .write_batch(|batch| {
3430 for i in 0..10u64 {
3431 let value = format!("value_{i}").into_bytes();
3432 batch.put::<TestTable>(i, &value)?;
3433 }
3434 Ok(())
3435 })
3436 .unwrap();
3437
3438 for i in 0..10u64 {
3440 let value = format!("value_{i}").into_bytes();
3441 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3442 }
3443
3444 provider
3446 .write_batch(|batch| {
3447 for i in 0..10u64 {
3448 batch.delete::<TestTable>(i)?;
3449 }
3450 Ok(())
3451 })
3452 .unwrap();
3453
3454 for i in 0..10u64 {
3456 assert_eq!(provider.get::<TestTable>(i).unwrap(), None);
3457 }
3458 }
3459
3460 #[test]
3461 fn test_with_real_table() {
3462 let temp_dir = TempDir::new().unwrap();
3463 let provider = RocksDBBuilder::new(temp_dir.path())
3464 .with_table::<tables::TransactionHashNumbers>()
3465 .with_metrics()
3466 .build()
3467 .unwrap();
3468
3469 let tx_hash = TxHash::from(B256::repeat_byte(1u8));
3470
3471 provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3473 assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3474
3475 provider
3477 .write_batch(|batch| {
3478 for i in 0..10u64 {
3479 let hash = TxHash::from(B256::repeat_byte(i as u8));
3480 let value = i * 100;
3481 batch.put::<tables::TransactionHashNumbers>(hash, &value)?;
3482 }
3483 Ok(())
3484 })
3485 .unwrap();
3486
3487 for i in 0..10u64 {
3489 let hash = TxHash::from(B256::repeat_byte(i as u8));
3490 assert_eq!(
3491 provider.get::<tables::TransactionHashNumbers>(hash).unwrap(),
3492 Some(i * 100)
3493 );
3494 }
3495 }
3496 #[test]
3497 fn test_statistics_enabled() {
3498 let temp_dir = TempDir::new().unwrap();
3499 let provider = RocksDBBuilder::new(temp_dir.path())
3501 .with_table::<TestTable>()
3502 .with_statistics()
3503 .build()
3504 .unwrap();
3505
3506 for i in 0..10 {
3508 let value = vec![i as u8];
3509 provider.put::<TestTable>(i, &value).unwrap();
3510 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3512 }
3513 }
3514
3515 #[test]
3519 fn test_statistics_level_skips_timers() {
3520 let temp_dir = TempDir::new().unwrap();
3521 let cache = Cache::new_lru_cache(1 << 20);
3522 let options = RocksDBBuilder::default_options(rocksdb::LogLevel::Info, &cache, true);
3523
3524 let db = DB::open(&options, temp_dir.path()).unwrap();
3525 for i in 0..10u8 {
3526 db.put([i], [i]).unwrap();
3527 assert_eq!(db.get([i]).unwrap(), Some(vec![i]));
3528 }
3529
3530 assert!(options.get_ticker_count(Ticker::NumberKeysRead) > 0);
3531 assert_eq!(options.get_histogram_data(Histogram::DbGet).count(), 0);
3532 }
3533
3534 #[test]
3535 fn test_data_persistence() {
3536 let temp_dir = TempDir::new().unwrap();
3537 let provider =
3538 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3539
3540 let value = vec![42u8; 1000];
3542 for i in 0..100 {
3543 provider.put::<TestTable>(i, &value).unwrap();
3544 }
3545
3546 for i in 0..100 {
3548 assert!(provider.get::<TestTable>(i).unwrap().is_some(), "Data should be readable");
3549 }
3550 }
3551
3552 #[test]
3553 fn test_transaction_read_your_writes() {
3554 let temp_dir = TempDir::new().unwrap();
3555 let provider =
3556 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3557
3558 let tx = provider.tx();
3560
3561 let key = 42u64;
3563 let value = b"test_value".to_vec();
3564 tx.put::<TestTable>(key, &value).unwrap();
3565
3566 let result = tx.get::<TestTable>(key).unwrap();
3568 assert_eq!(
3569 result,
3570 Some(value.clone()),
3571 "Transaction should see its own uncommitted writes"
3572 );
3573
3574 let provider_result = provider.get::<TestTable>(key).unwrap();
3576 assert_eq!(provider_result, None, "Uncommitted data should not be visible outside tx");
3577
3578 tx.commit().unwrap();
3580
3581 let committed_result = provider.get::<TestTable>(key).unwrap();
3583 assert_eq!(committed_result, Some(value), "Committed data should be visible");
3584 }
3585
3586 #[test]
3587 fn test_transaction_rollback() {
3588 let temp_dir = TempDir::new().unwrap();
3589 let provider =
3590 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3591
3592 let key = 100u64;
3594 let initial_value = b"initial".to_vec();
3595 provider.put::<TestTable>(key, &initial_value).unwrap();
3596
3597 let tx = provider.tx();
3599 let new_value = b"modified".to_vec();
3600 tx.put::<TestTable>(key, &new_value).unwrap();
3601
3602 assert_eq!(tx.get::<TestTable>(key).unwrap(), Some(new_value));
3604
3605 tx.rollback().unwrap();
3607
3608 let result = provider.get::<TestTable>(key).unwrap();
3610 assert_eq!(result, Some(initial_value), "Rollback should preserve original data");
3611 }
3612
3613 #[test]
3614 fn test_transaction_iterator() {
3615 let temp_dir = TempDir::new().unwrap();
3616 let provider =
3617 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3618
3619 let tx = provider.tx();
3621
3622 for i in 0..5u64 {
3624 let value = format!("value_{i}").into_bytes();
3625 tx.put::<TestTable>(i, &value).unwrap();
3626 }
3627
3628 let mut count = 0;
3630 for result in tx.iter::<TestTable>().unwrap() {
3631 let (key, value) = result.unwrap();
3632 assert_eq!(value, format!("value_{key}").into_bytes());
3633 count += 1;
3634 }
3635 assert_eq!(count, 5, "Iterator should see all uncommitted writes");
3636
3637 tx.commit().unwrap();
3639 }
3640
3641 #[test]
3642 fn test_batch_manual_commit() {
3643 let temp_dir = TempDir::new().unwrap();
3644 let provider =
3645 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3646
3647 let mut batch = provider.batch();
3649
3650 for i in 0..10u64 {
3652 let value = format!("batch_value_{i}").into_bytes();
3653 batch.put::<TestTable>(i, &value).unwrap();
3654 }
3655
3656 assert_eq!(batch.len(), 10);
3658 assert!(!batch.is_empty());
3659
3660 assert_eq!(provider.get::<TestTable>(0).unwrap(), None);
3662
3663 batch.commit().unwrap();
3665
3666 for i in 0..10u64 {
3668 let value = format!("batch_value_{i}").into_bytes();
3669 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3670 }
3671 }
3672
3673 #[test]
3674 fn test_first_and_last_entry() {
3675 let temp_dir = TempDir::new().unwrap();
3676 let provider =
3677 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3678
3679 assert_eq!(provider.first::<TestTable>().unwrap(), None);
3681 assert_eq!(provider.last::<TestTable>().unwrap(), None);
3682
3683 provider.put::<TestTable>(10, &b"value_10".to_vec()).unwrap();
3685 provider.put::<TestTable>(20, &b"value_20".to_vec()).unwrap();
3686 provider.put::<TestTable>(5, &b"value_5".to_vec()).unwrap();
3687
3688 let first = provider.first::<TestTable>().unwrap();
3690 assert_eq!(first, Some((5, b"value_5".to_vec())));
3691
3692 let last = provider.last::<TestTable>().unwrap();
3694 assert_eq!(last, Some((20, b"value_20".to_vec())));
3695 }
3696
3697 #[test]
3698 fn test_owned_history_snapshot_outlives_provider() {
3699 let temp_dir = TempDir::new().unwrap();
3700 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3701 let address = Address::repeat_byte(0x42);
3702 let slot = B256::repeat_byte(0x43);
3703 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3704 provider
3705 .put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &chunk)
3706 .unwrap();
3707 provider
3708 .put::<tables::StoragesHistory>(StorageShardedKey::new(address, slot, u64::MAX), &chunk)
3709 .unwrap();
3710
3711 let owned = provider.owned_snapshot();
3712 let snapshot = owned.as_snapshot();
3713 assert_eq!(
3714 snapshot.account_history_info(address, 200, None, u64::MAX).unwrap(),
3715 HistoryInfo::InChangeset(200)
3716 );
3717 assert_eq!(
3718 snapshot.storage_history_info(address, slot, 200, None, u64::MAX).unwrap(),
3719 HistoryInfo::InChangeset(200)
3720 );
3721 drop(provider);
3722
3723 std::thread::spawn(move || {
3725 let snapshot = owned.as_snapshot();
3726 for (block, expected) in [
3727 (400, HistoryInfo::InPlainState),
3728 (50, HistoryInfo::NotYetWritten),
3729 (200, HistoryInfo::InChangeset(200)),
3730 ] {
3731 assert_eq!(
3732 snapshot.account_history_info(address, block, None, u64::MAX).unwrap(),
3733 expected
3734 );
3735 assert_eq!(
3736 snapshot.storage_history_info(address, slot, block, None, u64::MAX).unwrap(),
3737 expected
3738 );
3739 }
3740 })
3741 .join()
3742 .unwrap();
3743 }
3744
3745 #[test]
3746 fn test_history_snapshot_cached_and_private_iterators_keep_same_view() {
3747 let temp_dir = TempDir::new().unwrap();
3748 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3749 let address = Address::repeat_byte(0x42);
3750 let slot = B256::repeat_byte(0x43);
3751 let account_key = ShardedKey::new(address, u64::MAX);
3752 let storage_key = StorageShardedKey::new(address, slot, u64::MAX);
3753 let chunk = IntegerList::new([100, 200]).unwrap();
3754 provider.put::<tables::AccountsHistory>(account_key.clone(), &chunk).unwrap();
3755 provider.put::<tables::StoragesHistory>(storage_key.clone(), &chunk).unwrap();
3756
3757 let owned = provider.owned_snapshot();
3758 let snapshot = owned.as_snapshot();
3759 let check = |snapshot: &RocksReadSnapshot<'_>, expected| {
3760 assert_eq!(
3761 snapshot.account_history_info(address, 125, None, u64::MAX).unwrap(),
3762 expected
3763 );
3764 assert_eq!(
3765 snapshot.storage_history_info(address, slot, 125, None, u64::MAX).unwrap(),
3766 expected
3767 );
3768 };
3769 check(snapshot, HistoryInfo::InChangeset(200));
3770
3771 let updated = IntegerList::new([100, 150, 200]).unwrap();
3772 provider.put::<tables::AccountsHistory>(account_key, &updated).unwrap();
3773 provider.put::<tables::StoragesHistory>(storage_key, &updated).unwrap();
3774 check(&provider.snapshot(), HistoryInfo::InChangeset(150));
3775 check(snapshot, HistoryInfo::InChangeset(200));
3776
3777 let _accounts = snapshot.accounts_history_iter.lock();
3779 let _storages = snapshot.storages_history_iter.lock();
3780 std::thread::scope(|scope| {
3781 scope.spawn(|| check(snapshot, HistoryInfo::InChangeset(200))).join().unwrap();
3782 });
3783 }
3784
3785 #[test]
3790 fn test_account_history_info_pruned_before_first_entry() {
3791 let temp_dir = TempDir::new().unwrap();
3792 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3793
3794 let address = Address::repeat_byte(0x42);
3795
3796 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3798 let shard_key = ShardedKey::new(address, u64::MAX);
3799 provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3800
3801 let result =
3806 provider.snapshot().account_history_info(address, 50, Some(100), u64::MAX).unwrap();
3807 assert_eq!(result, HistoryInfo::InChangeset(100));
3808 }
3809
3810 #[test]
3812 fn test_account_history_info_read_only_and_catch_up() {
3813 let temp_dir = TempDir::new().unwrap();
3814 let address = Address::repeat_byte(0x42);
3815 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3816 let shard_key = ShardedKey::new(address, u64::MAX);
3817
3818 let rw_provider =
3820 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3821 rw_provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3822
3823 let ro_provider = RocksDBBuilder::new(temp_dir.path())
3825 .with_default_tables()
3826 .with_read_only(true)
3827 .build()
3828 .unwrap();
3829
3830 let result =
3831 ro_provider.snapshot().account_history_info(address, 200, None, u64::MAX).unwrap();
3832 assert_eq!(result, HistoryInfo::InChangeset(200));
3833
3834 let result =
3835 ro_provider.snapshot().account_history_info(address, 50, None, u64::MAX).unwrap();
3836 assert_eq!(result, HistoryInfo::NotYetWritten);
3837
3838 let result =
3839 ro_provider.snapshot().account_history_info(address, 400, None, u64::MAX).unwrap();
3840 assert_eq!(result, HistoryInfo::InPlainState);
3841
3842 let address2 = Address::repeat_byte(0x43);
3844 let chunk2 = IntegerList::new([500, 600]).unwrap();
3845 let shard_key2 = ShardedKey::new(address2, u64::MAX);
3846 rw_provider.put::<tables::AccountsHistory>(shard_key2, &chunk2).unwrap();
3847
3848 let result =
3850 ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3851 assert_eq!(result, HistoryInfo::NotYetWritten);
3852
3853 ro_provider.try_catch_up_with_primary().unwrap();
3855
3856 let result =
3857 ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3858 assert_eq!(result, HistoryInfo::InChangeset(500));
3859 }
3860
3861 #[test]
3862 fn test_account_history_info_ignores_blocks_above_visible_tip() {
3863 let temp_dir = TempDir::new().unwrap();
3864 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3865
3866 let address = Address::repeat_byte(0x42);
3867
3868 provider
3869 .put::<tables::AccountsHistory>(
3870 ShardedKey::new(address, 110),
3871 &IntegerList::new([100, 110]).unwrap(),
3872 )
3873 .unwrap();
3874 provider
3875 .put::<tables::AccountsHistory>(
3876 ShardedKey::new(address, u64::MAX),
3877 &IntegerList::new([200, 210]).unwrap(),
3878 )
3879 .unwrap();
3880
3881 let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3882 assert_eq!(result, HistoryInfo::InPlainState);
3883 }
3884
3885 #[test]
3886 fn test_account_history_info_mixed_shard_respects_visible_tip() {
3887 let temp_dir = TempDir::new().unwrap();
3888 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3889
3890 let address = Address::repeat_byte(0x42);
3891 provider
3892 .put::<tables::AccountsHistory>(
3893 ShardedKey::new(address, u64::MAX),
3894 &IntegerList::new([100, 150, 300]).unwrap(),
3895 )
3896 .unwrap();
3897
3898 let result = provider.snapshot().account_history_info(address, 120, None, 200).unwrap();
3899 assert_eq!(result, HistoryInfo::InChangeset(150));
3900
3901 let result = provider.snapshot().account_history_info(address, 201, None, 200).unwrap();
3902 assert_eq!(result, HistoryInfo::InPlainState);
3903 }
3904
3905 #[test]
3906 fn test_account_history_info_only_stale_entries_use_fallback() {
3907 let temp_dir = TempDir::new().unwrap();
3908 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3909
3910 let address = Address::repeat_byte(0x42);
3911 provider
3912 .put::<tables::AccountsHistory>(
3913 ShardedKey::new(address, u64::MAX),
3914 &IntegerList::new([200, 210]).unwrap(),
3915 )
3916 .unwrap();
3917
3918 let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3919 assert_eq!(result, HistoryInfo::NotYetWritten);
3920
3921 let result =
3922 provider.snapshot().account_history_info(address, 150, Some(100), 150).unwrap();
3923 assert_eq!(result, HistoryInfo::MaybeInPlainState);
3924 }
3925
3926 #[test]
3927 fn test_account_history_shard_split_at_boundary() {
3928 let temp_dir = TempDir::new().unwrap();
3929 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3930
3931 let address = Address::repeat_byte(0x42);
3932 let limit = NUM_OF_INDICES_IN_SHARD;
3933
3934 let indices: Vec<u64> = (0..=(limit as u64)).collect();
3936 let mut batch = provider.batch();
3937 batch.append_account_history_shard(address, indices).unwrap();
3938 batch.commit().unwrap();
3939
3940 let completed_key = ShardedKey::new(address, (limit - 1) as u64);
3942 let sentinel_key = ShardedKey::new(address, u64::MAX);
3943
3944 let completed_shard = provider.get::<tables::AccountsHistory>(completed_key).unwrap();
3945 let sentinel_shard = provider.get::<tables::AccountsHistory>(sentinel_key).unwrap();
3946
3947 assert!(completed_shard.is_some(), "completed shard should exist");
3948 assert!(sentinel_shard.is_some(), "sentinel shard should exist");
3949
3950 let completed_shard = completed_shard.unwrap();
3951 let sentinel_shard = sentinel_shard.unwrap();
3952
3953 assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
3954 assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
3955 }
3956
3957 #[test]
3958 fn test_account_history_multiple_shard_splits() {
3959 let temp_dir = TempDir::new().unwrap();
3960 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3961
3962 let address = Address::repeat_byte(0x43);
3963 let limit = NUM_OF_INDICES_IN_SHARD;
3964
3965 let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
3967 let mut batch = provider.batch();
3968 batch.append_account_history_shard(address, first_batch_indices).unwrap();
3969 batch.commit().unwrap();
3970
3971 let sentinel_key = ShardedKey::new(address, u64::MAX);
3973 let shard = provider.get::<tables::AccountsHistory>(sentinel_key.clone()).unwrap();
3974 assert!(shard.is_some());
3975 assert_eq!(shard.unwrap().len(), limit as u64);
3976
3977 let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
3979 let mut batch = provider.batch();
3980 batch.append_account_history_shard(address, second_batch_indices).unwrap();
3981 batch.commit().unwrap();
3982
3983 let first_completed = ShardedKey::new(address, (limit - 1) as u64);
3985 let second_completed = ShardedKey::new(address, (2 * limit - 1) as u64);
3986
3987 assert!(
3988 provider.get::<tables::AccountsHistory>(first_completed).unwrap().is_some(),
3989 "first completed shard should exist"
3990 );
3991 assert!(
3992 provider.get::<tables::AccountsHistory>(second_completed).unwrap().is_some(),
3993 "second completed shard should exist"
3994 );
3995 assert!(
3996 provider.get::<tables::AccountsHistory>(sentinel_key).unwrap().is_some(),
3997 "sentinel shard should exist"
3998 );
3999 }
4000
4001 #[test]
4002 fn test_storage_history_shard_split_at_boundary() {
4003 let temp_dir = TempDir::new().unwrap();
4004 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4005
4006 let address = Address::repeat_byte(0x44);
4007 let slot = B256::repeat_byte(0x55);
4008 let limit = NUM_OF_INDICES_IN_SHARD;
4009
4010 let indices: Vec<u64> = (0..=(limit as u64)).collect();
4012 let mut batch = provider.batch();
4013 batch.append_storage_history_shard(address, slot, indices).unwrap();
4014 batch.commit().unwrap();
4015
4016 let completed_key = StorageShardedKey::new(address, slot, (limit - 1) as u64);
4018 let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
4019
4020 let completed_shard = provider.get::<tables::StoragesHistory>(completed_key).unwrap();
4021 let sentinel_shard = provider.get::<tables::StoragesHistory>(sentinel_key).unwrap();
4022
4023 assert!(completed_shard.is_some(), "completed shard should exist");
4024 assert!(sentinel_shard.is_some(), "sentinel shard should exist");
4025
4026 let completed_shard = completed_shard.unwrap();
4027 let sentinel_shard = sentinel_shard.unwrap();
4028
4029 assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
4030 assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
4031 }
4032
4033 #[test]
4034 fn test_storage_history_multiple_shard_splits() {
4035 let temp_dir = TempDir::new().unwrap();
4036 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4037
4038 let address = Address::repeat_byte(0x46);
4039 let slot = B256::repeat_byte(0x57);
4040 let limit = NUM_OF_INDICES_IN_SHARD;
4041
4042 let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
4044 let mut batch = provider.batch();
4045 batch.append_storage_history_shard(address, slot, first_batch_indices).unwrap();
4046 batch.commit().unwrap();
4047
4048 let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
4050 let shard = provider.get::<tables::StoragesHistory>(sentinel_key.clone()).unwrap();
4051 assert!(shard.is_some());
4052 assert_eq!(shard.unwrap().len(), limit as u64);
4053
4054 let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
4056 let mut batch = provider.batch();
4057 batch.append_storage_history_shard(address, slot, second_batch_indices).unwrap();
4058 batch.commit().unwrap();
4059
4060 let first_completed = StorageShardedKey::new(address, slot, (limit - 1) as u64);
4062 let second_completed = StorageShardedKey::new(address, slot, (2 * limit - 1) as u64);
4063
4064 assert!(
4065 provider.get::<tables::StoragesHistory>(first_completed).unwrap().is_some(),
4066 "first completed shard should exist"
4067 );
4068 assert!(
4069 provider.get::<tables::StoragesHistory>(second_completed).unwrap().is_some(),
4070 "second completed shard should exist"
4071 );
4072 assert!(
4073 provider.get::<tables::StoragesHistory>(sentinel_key).unwrap().is_some(),
4074 "sentinel shard should exist"
4075 );
4076 }
4077
4078 #[test]
4079 fn test_clear_table() {
4080 let temp_dir = TempDir::new().unwrap();
4081 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4082
4083 let address = Address::repeat_byte(0x42);
4084 let key = ShardedKey::new(address, u64::MAX);
4085 let blocks = BlockNumberList::new_pre_sorted([1, 2, 3]);
4086
4087 provider.put::<tables::AccountsHistory>(key.clone(), &blocks).unwrap();
4088 assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
4089
4090 provider.clear::<tables::AccountsHistory>().unwrap();
4091
4092 assert!(
4093 provider.get::<tables::AccountsHistory>(key).unwrap().is_none(),
4094 "table should be empty after clear"
4095 );
4096 assert!(
4097 provider.first::<tables::AccountsHistory>().unwrap().is_none(),
4098 "first() should return None after clear"
4099 );
4100 }
4101
4102 #[test]
4103 fn test_clear_empty_table() {
4104 let temp_dir = TempDir::new().unwrap();
4105 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4106
4107 assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
4108
4109 provider.clear::<tables::AccountsHistory>().unwrap();
4110
4111 assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
4112 }
4113
4114 #[test]
4115 fn test_unwind_account_history_to_basic() {
4116 let temp_dir = TempDir::new().unwrap();
4117 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4118
4119 let address = Address::repeat_byte(0x42);
4120
4121 let mut batch = provider.batch();
4123 batch.append_account_history_shard(address, 0..=10).unwrap();
4124 batch.commit().unwrap();
4125
4126 let key = ShardedKey::new(address, u64::MAX);
4128 let result = provider.get::<tables::AccountsHistory>(key.clone()).unwrap();
4129 assert!(result.is_some());
4130 let blocks: Vec<u64> = result.unwrap().iter().collect();
4131 assert_eq!(blocks, (0..=10).collect::<Vec<_>>());
4132
4133 let mut batch = provider.batch();
4135 batch.unwind_account_history_to(address, 5).unwrap();
4136 batch.commit().unwrap();
4137
4138 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4140 assert!(result.is_some());
4141 let blocks: Vec<u64> = result.unwrap().iter().collect();
4142 assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
4143 }
4144
4145 #[test]
4146 fn test_unwind_account_history_to_removes_all() {
4147 let temp_dir = TempDir::new().unwrap();
4148 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4149
4150 let address = Address::repeat_byte(0x42);
4151
4152 let mut batch = provider.batch();
4154 batch.append_account_history_shard(address, 5..=10).unwrap();
4155 batch.commit().unwrap();
4156
4157 let mut batch = provider.batch();
4159 batch.unwind_account_history_to(address, 4).unwrap();
4160 batch.commit().unwrap();
4161
4162 let key = ShardedKey::new(address, u64::MAX);
4164 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4165 assert!(result.is_none(), "Should have no data after full unwind");
4166 }
4167
4168 #[test]
4169 fn test_unwind_account_history_to_no_op() {
4170 let temp_dir = TempDir::new().unwrap();
4171 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4172
4173 let address = Address::repeat_byte(0x42);
4174
4175 let mut batch = provider.batch();
4177 batch.append_account_history_shard(address, 0..=5).unwrap();
4178 batch.commit().unwrap();
4179
4180 let mut batch = provider.batch();
4182 batch.unwind_account_history_to(address, 10).unwrap();
4183 batch.commit().unwrap();
4184
4185 let key = ShardedKey::new(address, u64::MAX);
4187 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4188 assert!(result.is_some());
4189 let blocks: Vec<u64> = result.unwrap().iter().collect();
4190 assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
4191 }
4192
4193 #[test]
4194 fn test_unwind_account_history_to_block_zero() {
4195 let temp_dir = TempDir::new().unwrap();
4196 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4197
4198 let address = Address::repeat_byte(0x42);
4199
4200 let mut batch = provider.batch();
4202 batch.append_account_history_shard(address, 0..=5).unwrap();
4203 batch.commit().unwrap();
4204
4205 let mut batch = provider.batch();
4208 batch.unwind_account_history_to(address, 0).unwrap();
4209 batch.commit().unwrap();
4210
4211 let key = ShardedKey::new(address, u64::MAX);
4213 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4214 assert!(result.is_some());
4215 let blocks: Vec<u64> = result.unwrap().iter().collect();
4216 assert_eq!(blocks, vec![0]);
4217 }
4218
4219 #[test]
4220 fn test_unwind_account_history_to_multi_shard() {
4221 let temp_dir = TempDir::new().unwrap();
4222 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4223
4224 let address = Address::repeat_byte(0x42);
4225
4226 let mut batch = provider.batch();
4229
4230 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4232 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4233
4234 let shard2 = BlockNumberList::new_pre_sorted(51..=100);
4236 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
4237
4238 batch.commit().unwrap();
4239
4240 let shards = provider.account_history_shards(address).unwrap();
4242 assert_eq!(shards.len(), 2);
4243
4244 let mut batch = provider.batch();
4246 batch.unwind_account_history_to(address, 75).unwrap();
4247 batch.commit().unwrap();
4248
4249 let shards = provider.account_history_shards(address).unwrap();
4251 assert_eq!(shards.len(), 2);
4252
4253 assert_eq!(shards[0].0.highest_block_number, 50);
4255 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4256
4257 assert_eq!(shards[1].0.highest_block_number, u64::MAX);
4259 assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
4260 }
4261
4262 #[test]
4263 fn test_unwind_account_history_to_multi_shard_boundary_empty() {
4264 let temp_dir = TempDir::new().unwrap();
4265 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4266
4267 let address = Address::repeat_byte(0x42);
4268
4269 let mut batch = provider.batch();
4271
4272 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4274 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4275
4276 let shard2 = BlockNumberList::new_pre_sorted(75..=100);
4278 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
4279
4280 batch.commit().unwrap();
4281
4282 let mut batch = provider.batch();
4284 batch.unwind_account_history_to(address, 60).unwrap();
4285 batch.commit().unwrap();
4286
4287 let shards = provider.account_history_shards(address).unwrap();
4289 assert_eq!(shards.len(), 1);
4290 assert_eq!(shards[0].0.highest_block_number, u64::MAX);
4291 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4292 }
4293
4294 #[test]
4295 fn test_account_history_shards_iterator() {
4296 let temp_dir = TempDir::new().unwrap();
4297 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4298
4299 let address = Address::repeat_byte(0x42);
4300 let other_address = Address::repeat_byte(0x43);
4301
4302 let mut batch = provider.batch();
4304 batch.append_account_history_shard(address, 0..=5).unwrap();
4305 batch.append_account_history_shard(other_address, 10..=15).unwrap();
4306 batch.commit().unwrap();
4307
4308 let shards = provider.account_history_shards(address).unwrap();
4310 assert_eq!(shards.len(), 1);
4311 assert_eq!(shards[0].0.key, address);
4312
4313 let shards = provider.account_history_shards(other_address).unwrap();
4315 assert_eq!(shards.len(), 1);
4316 assert_eq!(shards[0].0.key, other_address);
4317
4318 let non_existent = Address::repeat_byte(0x99);
4320 let shards = provider.account_history_shards(non_existent).unwrap();
4321 assert!(shards.is_empty());
4322 }
4323
4324 #[test]
4325 fn test_clear_account_history() {
4326 let temp_dir = TempDir::new().unwrap();
4327 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4328
4329 let address = Address::repeat_byte(0x42);
4330
4331 let mut batch = provider.batch();
4333 batch.append_account_history_shard(address, 0..=10).unwrap();
4334 batch.commit().unwrap();
4335
4336 let mut batch = provider.batch();
4338 batch.clear_account_history(address).unwrap();
4339 batch.commit().unwrap();
4340
4341 let shards = provider.account_history_shards(address).unwrap();
4343 assert!(shards.is_empty(), "All shards should be deleted");
4344 }
4345
4346 #[test]
4347 fn test_unwind_non_sentinel_boundary() {
4348 let temp_dir = TempDir::new().unwrap();
4349 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4350
4351 let address = Address::repeat_byte(0x42);
4352
4353 let mut batch = provider.batch();
4355
4356 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4358 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4359
4360 let shard2 = BlockNumberList::new_pre_sorted(51..=100);
4362 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 100), &shard2).unwrap();
4363
4364 let shard3 = BlockNumberList::new_pre_sorted(101..=150);
4366 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard3).unwrap();
4367
4368 batch.commit().unwrap();
4369
4370 let shards = provider.account_history_shards(address).unwrap();
4372 assert_eq!(shards.len(), 3);
4373
4374 let mut batch = provider.batch();
4376 batch.unwind_account_history_to(address, 75).unwrap();
4377 batch.commit().unwrap();
4378
4379 let shards = provider.account_history_shards(address).unwrap();
4381 assert_eq!(shards.len(), 2);
4382
4383 assert_eq!(shards[0].0.highest_block_number, 50);
4385 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4386
4387 assert_eq!(shards[1].0.highest_block_number, u64::MAX);
4389 assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
4390 }
4391
4392 #[test]
4393 fn test_batch_auto_commit_on_threshold() {
4394 let temp_dir = TempDir::new().unwrap();
4395 let provider =
4396 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
4397
4398 let mut batch = RocksDBBatch {
4400 provider: &provider,
4401 inner: WriteBatchWithTransaction::<true>::default(),
4402 buf: Vec::new(),
4403 auto_commit_threshold: Some(1024), };
4405
4406 for i in 0..100u64 {
4409 let value = format!("value_{i:04}").into_bytes();
4410 batch.put::<TestTable>(i, &value).unwrap();
4411 }
4412
4413 let first_visible = provider.get::<TestTable>(0).unwrap();
4416 assert!(first_visible.is_some(), "Auto-committed data should be visible");
4417
4418 batch.commit().unwrap();
4420
4421 for i in 0..100u64 {
4423 let value = format!("value_{i:04}").into_bytes();
4424 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
4425 }
4426 }
4427
4428 struct AccountPruneCase {
4432 name: &'static str,
4433 initial_shards: &'static [(u64, &'static [u64])],
4434 prune_to: u64,
4435 expected_outcome: PruneShardOutcome,
4436 expected_shards: &'static [(u64, &'static [u64])],
4437 }
4438
4439 struct StoragePruneCase {
4441 name: &'static str,
4442 initial_shards: &'static [(u64, &'static [u64])],
4443 prune_to: u64,
4444 expected_outcome: PruneShardOutcome,
4445 expected_shards: &'static [(u64, &'static [u64])],
4446 }
4447
4448 #[test]
4449 fn test_prune_account_history_cases() {
4450 const MAX: u64 = u64::MAX;
4451 const CASES: &[AccountPruneCase] = &[
4452 AccountPruneCase {
4453 name: "single_shard_truncate",
4454 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4455 prune_to: 25,
4456 expected_outcome: PruneShardOutcome::Updated,
4457 expected_shards: &[(MAX, &[30, 40])],
4458 },
4459 AccountPruneCase {
4460 name: "single_shard_delete_all",
4461 initial_shards: &[(MAX, &[10, 20])],
4462 prune_to: 20,
4463 expected_outcome: PruneShardOutcome::Deleted,
4464 expected_shards: &[],
4465 },
4466 AccountPruneCase {
4467 name: "single_shard_noop",
4468 initial_shards: &[(MAX, &[10, 20])],
4469 prune_to: 5,
4470 expected_outcome: PruneShardOutcome::Unchanged,
4471 expected_shards: &[(MAX, &[10, 20])],
4472 },
4473 AccountPruneCase {
4474 name: "no_shards",
4475 initial_shards: &[],
4476 prune_to: 100,
4477 expected_outcome: PruneShardOutcome::Unchanged,
4478 expected_shards: &[],
4479 },
4480 AccountPruneCase {
4481 name: "multi_shard_truncate_first",
4482 initial_shards: &[(30, &[10, 20, 30]), (MAX, &[40, 50, 60])],
4483 prune_to: 25,
4484 expected_outcome: PruneShardOutcome::Updated,
4485 expected_shards: &[(30, &[30]), (MAX, &[40, 50, 60])],
4486 },
4487 AccountPruneCase {
4488 name: "delete_first_shard_sentinel_unchanged",
4489 initial_shards: &[(20, &[10, 20]), (MAX, &[30, 40])],
4490 prune_to: 20,
4491 expected_outcome: PruneShardOutcome::Deleted,
4492 expected_shards: &[(MAX, &[30, 40])],
4493 },
4494 AccountPruneCase {
4495 name: "multi_shard_delete_all_but_last",
4496 initial_shards: &[(10, &[5, 10]), (20, &[15, 20]), (MAX, &[25, 30])],
4497 prune_to: 22,
4498 expected_outcome: PruneShardOutcome::Deleted,
4499 expected_shards: &[(MAX, &[25, 30])],
4500 },
4501 AccountPruneCase {
4502 name: "mid_shard_preserves_key",
4503 initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4504 prune_to: 25,
4505 expected_outcome: PruneShardOutcome::Updated,
4506 expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4507 },
4508 AccountPruneCase {
4510 name: "equiv_delete_early_shards_keep_sentinel",
4511 initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4512 prune_to: 55,
4513 expected_outcome: PruneShardOutcome::Deleted,
4514 expected_shards: &[(MAX, &[60, 70])],
4515 },
4516 AccountPruneCase {
4517 name: "equiv_sentinel_becomes_empty_with_prev",
4518 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4519 prune_to: 40,
4520 expected_outcome: PruneShardOutcome::Deleted,
4521 expected_shards: &[(MAX, &[50])],
4522 },
4523 AccountPruneCase {
4524 name: "equiv_all_shards_become_empty",
4525 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4526 prune_to: 51,
4527 expected_outcome: PruneShardOutcome::Deleted,
4528 expected_shards: &[],
4529 },
4530 AccountPruneCase {
4531 name: "equiv_non_sentinel_last_shard_promoted",
4532 initial_shards: &[(100, &[50, 75, 100])],
4533 prune_to: 60,
4534 expected_outcome: PruneShardOutcome::Updated,
4535 expected_shards: &[(MAX, &[75, 100])],
4536 },
4537 AccountPruneCase {
4538 name: "equiv_filter_within_shard",
4539 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4540 prune_to: 25,
4541 expected_outcome: PruneShardOutcome::Updated,
4542 expected_shards: &[(MAX, &[30, 40])],
4543 },
4544 AccountPruneCase {
4545 name: "equiv_multi_shard_partial_delete",
4546 initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4547 prune_to: 35,
4548 expected_outcome: PruneShardOutcome::Deleted,
4549 expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4550 },
4551 ];
4552
4553 let address = Address::repeat_byte(0x42);
4554
4555 for case in CASES {
4556 let temp_dir = TempDir::new().unwrap();
4557 let provider =
4558 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4559
4560 let mut batch = provider.batch();
4562 for (highest, blocks) in case.initial_shards {
4563 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4564 batch
4565 .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4566 .unwrap();
4567 }
4568 batch.commit().unwrap();
4569
4570 let mut batch = provider.batch();
4572 let outcome = batch.prune_account_history_to(address, case.prune_to).unwrap();
4573 batch.commit().unwrap();
4574
4575 assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4577
4578 let shards = provider.account_history_shards(address).unwrap();
4580 assert_eq!(
4581 shards.len(),
4582 case.expected_shards.len(),
4583 "case '{}': wrong shard count",
4584 case.name
4585 );
4586 for (i, ((key, blocks), (exp_key, exp_blocks))) in
4587 shards.iter().zip(case.expected_shards.iter()).enumerate()
4588 {
4589 assert_eq!(
4590 key.highest_block_number, *exp_key,
4591 "case '{}': shard {} wrong key",
4592 case.name, i
4593 );
4594 assert_eq!(
4595 blocks.iter().collect::<Vec<_>>(),
4596 *exp_blocks,
4597 "case '{}': shard {} wrong blocks",
4598 case.name,
4599 i
4600 );
4601 }
4602 }
4603 }
4604
4605 #[test]
4606 fn test_prune_storage_history_cases() {
4607 const MAX: u64 = u64::MAX;
4608 const CASES: &[StoragePruneCase] = &[
4609 StoragePruneCase {
4610 name: "single_shard_truncate",
4611 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4612 prune_to: 25,
4613 expected_outcome: PruneShardOutcome::Updated,
4614 expected_shards: &[(MAX, &[30, 40])],
4615 },
4616 StoragePruneCase {
4617 name: "single_shard_delete_all",
4618 initial_shards: &[(MAX, &[10, 20])],
4619 prune_to: 20,
4620 expected_outcome: PruneShardOutcome::Deleted,
4621 expected_shards: &[],
4622 },
4623 StoragePruneCase {
4624 name: "noop",
4625 initial_shards: &[(MAX, &[10, 20])],
4626 prune_to: 5,
4627 expected_outcome: PruneShardOutcome::Unchanged,
4628 expected_shards: &[(MAX, &[10, 20])],
4629 },
4630 StoragePruneCase {
4631 name: "no_shards",
4632 initial_shards: &[],
4633 prune_to: 100,
4634 expected_outcome: PruneShardOutcome::Unchanged,
4635 expected_shards: &[],
4636 },
4637 StoragePruneCase {
4638 name: "mid_shard_preserves_key",
4639 initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4640 prune_to: 25,
4641 expected_outcome: PruneShardOutcome::Updated,
4642 expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4643 },
4644 StoragePruneCase {
4646 name: "equiv_sentinel_promotion",
4647 initial_shards: &[(100, &[50, 75, 100])],
4648 prune_to: 60,
4649 expected_outcome: PruneShardOutcome::Updated,
4650 expected_shards: &[(MAX, &[75, 100])],
4651 },
4652 StoragePruneCase {
4653 name: "equiv_delete_early_shards_keep_sentinel",
4654 initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4655 prune_to: 55,
4656 expected_outcome: PruneShardOutcome::Deleted,
4657 expected_shards: &[(MAX, &[60, 70])],
4658 },
4659 StoragePruneCase {
4660 name: "equiv_sentinel_becomes_empty_with_prev",
4661 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4662 prune_to: 40,
4663 expected_outcome: PruneShardOutcome::Deleted,
4664 expected_shards: &[(MAX, &[50])],
4665 },
4666 StoragePruneCase {
4667 name: "equiv_all_shards_become_empty",
4668 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4669 prune_to: 51,
4670 expected_outcome: PruneShardOutcome::Deleted,
4671 expected_shards: &[],
4672 },
4673 StoragePruneCase {
4674 name: "equiv_filter_within_shard",
4675 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4676 prune_to: 25,
4677 expected_outcome: PruneShardOutcome::Updated,
4678 expected_shards: &[(MAX, &[30, 40])],
4679 },
4680 StoragePruneCase {
4681 name: "equiv_multi_shard_partial_delete",
4682 initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4683 prune_to: 35,
4684 expected_outcome: PruneShardOutcome::Deleted,
4685 expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4686 },
4687 ];
4688
4689 let address = Address::repeat_byte(0x42);
4690 let storage_key = B256::repeat_byte(0x01);
4691
4692 for case in CASES {
4693 let temp_dir = TempDir::new().unwrap();
4694 let provider =
4695 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4696
4697 let mut batch = provider.batch();
4699 for (highest, blocks) in case.initial_shards {
4700 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4701 let key = if *highest == MAX {
4702 StorageShardedKey::last(address, storage_key)
4703 } else {
4704 StorageShardedKey::new(address, storage_key, *highest)
4705 };
4706 batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4707 }
4708 batch.commit().unwrap();
4709
4710 let mut batch = provider.batch();
4712 let outcome =
4713 batch.prune_storage_history_to(address, storage_key, case.prune_to).unwrap();
4714 batch.commit().unwrap();
4715
4716 assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4718
4719 let shards = provider.storage_history_shards(address, storage_key).unwrap();
4721 assert_eq!(
4722 shards.len(),
4723 case.expected_shards.len(),
4724 "case '{}': wrong shard count",
4725 case.name
4726 );
4727 for (i, ((key, blocks), (exp_key, exp_blocks))) in
4728 shards.iter().zip(case.expected_shards.iter()).enumerate()
4729 {
4730 assert_eq!(
4731 key.sharded_key.highest_block_number, *exp_key,
4732 "case '{}': shard {} wrong key",
4733 case.name, i
4734 );
4735 assert_eq!(
4736 blocks.iter().collect::<Vec<_>>(),
4737 *exp_blocks,
4738 "case '{}': shard {} wrong blocks",
4739 case.name,
4740 i
4741 );
4742 }
4743 }
4744 }
4745
4746 #[test]
4747 fn test_prune_storage_history_does_not_affect_other_slots() {
4748 let temp_dir = TempDir::new().unwrap();
4749 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4750
4751 let address = Address::repeat_byte(0x42);
4752 let slot1 = B256::repeat_byte(0x01);
4753 let slot2 = B256::repeat_byte(0x02);
4754
4755 let mut batch = provider.batch();
4757 batch
4758 .put::<tables::StoragesHistory>(
4759 StorageShardedKey::last(address, slot1),
4760 &BlockNumberList::new_pre_sorted([10u64, 20]),
4761 )
4762 .unwrap();
4763 batch
4764 .put::<tables::StoragesHistory>(
4765 StorageShardedKey::last(address, slot2),
4766 &BlockNumberList::new_pre_sorted([30u64, 40]),
4767 )
4768 .unwrap();
4769 batch.commit().unwrap();
4770
4771 let mut batch = provider.batch();
4773 let outcome = batch.prune_storage_history_to(address, slot1, 20).unwrap();
4774 batch.commit().unwrap();
4775
4776 assert_eq!(outcome, PruneShardOutcome::Deleted);
4777
4778 let shards1 = provider.storage_history_shards(address, slot1).unwrap();
4780 assert!(shards1.is_empty());
4781
4782 let shards2 = provider.storage_history_shards(address, slot2).unwrap();
4784 assert_eq!(shards2.len(), 1);
4785 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![30, 40]);
4786 }
4787
4788 #[test]
4789 fn test_prune_invariants() {
4790 let address = Address::repeat_byte(0x42);
4792 let storage_key = B256::repeat_byte(0x01);
4793
4794 #[expect(clippy::type_complexity)]
4796 let invariant_cases: &[(&[(u64, &[u64])], u64)] = &[
4797 (&[(10, &[5, 10]), (20, &[15, 20]), (u64::MAX, &[25, 30])], 20),
4799 (&[(100, &[50, 100])], 60),
4801 ];
4802
4803 for (initial_shards, prune_to) in invariant_cases {
4804 {
4806 let temp_dir = TempDir::new().unwrap();
4807 let provider =
4808 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4809
4810 let mut batch = provider.batch();
4811 for (highest, blocks) in *initial_shards {
4812 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4813 batch
4814 .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4815 .unwrap();
4816 }
4817 batch.commit().unwrap();
4818
4819 let mut batch = provider.batch();
4820 batch.prune_account_history_to(address, *prune_to).unwrap();
4821 batch.commit().unwrap();
4822
4823 let shards = provider.account_history_shards(address).unwrap();
4824
4825 for (key, blocks) in &shards {
4827 assert!(
4828 !blocks.is_empty(),
4829 "Account: empty shard at key {}",
4830 key.highest_block_number
4831 );
4832 }
4833
4834 if !shards.is_empty() {
4836 let last = shards.last().unwrap();
4837 assert_eq!(
4838 last.0.highest_block_number,
4839 u64::MAX,
4840 "Account: last shard must be sentinel"
4841 );
4842 }
4843 }
4844
4845 {
4847 let temp_dir = TempDir::new().unwrap();
4848 let provider =
4849 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4850
4851 let mut batch = provider.batch();
4852 for (highest, blocks) in *initial_shards {
4853 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4854 let key = if *highest == u64::MAX {
4855 StorageShardedKey::last(address, storage_key)
4856 } else {
4857 StorageShardedKey::new(address, storage_key, *highest)
4858 };
4859 batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4860 }
4861 batch.commit().unwrap();
4862
4863 let mut batch = provider.batch();
4864 batch.prune_storage_history_to(address, storage_key, *prune_to).unwrap();
4865 batch.commit().unwrap();
4866
4867 let shards = provider.storage_history_shards(address, storage_key).unwrap();
4868
4869 for (key, blocks) in &shards {
4871 assert!(
4872 !blocks.is_empty(),
4873 "Storage: empty shard at key {}",
4874 key.sharded_key.highest_block_number
4875 );
4876 }
4877
4878 if !shards.is_empty() {
4880 let last = shards.last().unwrap();
4881 assert_eq!(
4882 last.0.sharded_key.highest_block_number,
4883 u64::MAX,
4884 "Storage: last shard must be sentinel"
4885 );
4886 }
4887 }
4888 }
4889 }
4890
4891 #[test]
4892 fn test_prune_account_history_batch_multiple_sorted_targets() {
4893 let temp_dir = TempDir::new().unwrap();
4894 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4895
4896 let addr1 = Address::repeat_byte(0x01);
4897 let addr2 = Address::repeat_byte(0x02);
4898 let addr3 = Address::repeat_byte(0x03);
4899
4900 let mut batch = provider.batch();
4902 batch
4903 .put::<tables::AccountsHistory>(
4904 ShardedKey::new(addr1, u64::MAX),
4905 &BlockNumberList::new_pre_sorted([10, 20, 30]),
4906 )
4907 .unwrap();
4908 batch
4909 .put::<tables::AccountsHistory>(
4910 ShardedKey::new(addr2, u64::MAX),
4911 &BlockNumberList::new_pre_sorted([5, 10, 15]),
4912 )
4913 .unwrap();
4914 batch
4915 .put::<tables::AccountsHistory>(
4916 ShardedKey::new(addr3, u64::MAX),
4917 &BlockNumberList::new_pre_sorted([100, 200]),
4918 )
4919 .unwrap();
4920 batch.commit().unwrap();
4921
4922 let mut targets = vec![(addr1, 15), (addr2, 10), (addr3, 50)];
4924 targets.sort_by_key(|(addr, _)| *addr);
4925
4926 let mut batch = provider.batch();
4927 let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4928 batch.commit().unwrap();
4929
4930 assert_eq!(outcomes.updated, 2);
4934 assert_eq!(outcomes.unchanged, 1);
4935
4936 let shards1 = provider.account_history_shards(addr1).unwrap();
4937 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
4938
4939 let shards2 = provider.account_history_shards(addr2).unwrap();
4940 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15]);
4941
4942 let shards3 = provider.account_history_shards(addr3).unwrap();
4943 assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![100, 200]);
4944 }
4945
4946 #[test]
4947 fn test_prune_account_history_batch_target_with_no_shards() {
4948 let temp_dir = TempDir::new().unwrap();
4949 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4950
4951 let addr1 = Address::repeat_byte(0x01);
4952 let addr2 = Address::repeat_byte(0x02); let addr3 = Address::repeat_byte(0x03);
4954
4955 let mut batch = provider.batch();
4957 batch
4958 .put::<tables::AccountsHistory>(
4959 ShardedKey::new(addr1, u64::MAX),
4960 &BlockNumberList::new_pre_sorted([10, 20]),
4961 )
4962 .unwrap();
4963 batch
4964 .put::<tables::AccountsHistory>(
4965 ShardedKey::new(addr3, u64::MAX),
4966 &BlockNumberList::new_pre_sorted([30, 40]),
4967 )
4968 .unwrap();
4969 batch.commit().unwrap();
4970
4971 let mut targets = vec![(addr1, 15), (addr2, 100), (addr3, 35)];
4973 targets.sort_by_key(|(addr, _)| *addr);
4974
4975 let mut batch = provider.batch();
4976 let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4977 batch.commit().unwrap();
4978
4979 assert_eq!(outcomes.updated, 2);
4983 assert_eq!(outcomes.unchanged, 1);
4984
4985 let shards1 = provider.account_history_shards(addr1).unwrap();
4986 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20]);
4987
4988 let shards3 = provider.account_history_shards(addr3).unwrap();
4989 assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![40]);
4990 }
4991
4992 #[test]
4993 fn test_prune_storage_history_batch_multiple_sorted_targets() {
4994 let temp_dir = TempDir::new().unwrap();
4995 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4996
4997 let addr = Address::repeat_byte(0x42);
4998 let slot1 = B256::repeat_byte(0x01);
4999 let slot2 = B256::repeat_byte(0x02);
5000
5001 let mut batch = provider.batch();
5003 batch
5004 .put::<tables::StoragesHistory>(
5005 StorageShardedKey::new(addr, slot1, u64::MAX),
5006 &BlockNumberList::new_pre_sorted([10, 20, 30]),
5007 )
5008 .unwrap();
5009 batch
5010 .put::<tables::StoragesHistory>(
5011 StorageShardedKey::new(addr, slot2, u64::MAX),
5012 &BlockNumberList::new_pre_sorted([5, 15, 25]),
5013 )
5014 .unwrap();
5015 batch.commit().unwrap();
5016
5017 let mut targets = vec![((addr, slot1), 15), ((addr, slot2), 10)];
5019 targets.sort_by_key(|((a, s), _)| (*a, *s));
5020
5021 let mut batch = provider.batch();
5022 let outcomes = batch.prune_storage_history_batch(&targets).unwrap();
5023 batch.commit().unwrap();
5024
5025 assert_eq!(outcomes.updated, 2);
5026
5027 let shards1 = provider.storage_history_shards(addr, slot1).unwrap();
5028 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
5029
5030 let shards2 = provider.storage_history_shards(addr, slot2).unwrap();
5031 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15, 25]);
5032 }
5033
5034 fn account_shard_layout(provider: &RocksDBProvider, address: Address) -> Vec<(u64, Vec<u64>)> {
5036 provider
5037 .account_history_shards(address)
5038 .unwrap()
5039 .into_iter()
5040 .map(|(key, list)| (key.highest_block_number, list.iter().collect::<Vec<_>>()))
5041 .collect()
5042 }
5043
5044 fn storage_shard_layout(
5046 provider: &RocksDBProvider,
5047 address: Address,
5048 storage_key: B256,
5049 ) -> Vec<(u64, Vec<u64>)> {
5050 provider
5051 .storage_history_shards(address, storage_key)
5052 .unwrap()
5053 .into_iter()
5054 .map(|(key, list)| {
5055 (key.sharded_key.highest_block_number, list.iter().collect::<Vec<_>>())
5056 })
5057 .collect()
5058 }
5059
5060 fn seed_three_storage_shards(provider: &RocksDBProvider, address: Address, storage_key: B256) {
5061 let mut batch = provider.batch();
5062 batch
5063 .put::<tables::StoragesHistory>(
5064 StorageShardedKey::new(address, storage_key, 100),
5065 &BlockNumberList::new_pre_sorted([10, 50, 100]),
5066 )
5067 .unwrap();
5068 batch
5069 .put::<tables::StoragesHistory>(
5070 StorageShardedKey::new(address, storage_key, 200),
5071 &BlockNumberList::new_pre_sorted([150, 200]),
5072 )
5073 .unwrap();
5074 batch
5075 .put::<tables::StoragesHistory>(
5076 StorageShardedKey::last(address, storage_key),
5077 &BlockNumberList::new_pre_sorted([250, 300]),
5078 )
5079 .unwrap();
5080 batch.commit().unwrap();
5081 }
5082
5083 #[test]
5084 fn test_prune_storage_history_batch_trims_sentinel_once_earlier_shards_expire() {
5085 let temp_dir = TempDir::new().unwrap();
5086 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
5087
5088 let addr = Address::repeat_byte(0x42);
5089 let slot = B256::repeat_byte(0x01);
5090 seed_three_storage_shards(&provider, addr, slot);
5091
5092 let mut batch = provider.batch();
5094 let outcomes = batch.prune_storage_history_batch(&[((addr, slot), 250)]).unwrap();
5095 batch.commit().unwrap();
5096
5097 assert_eq!(outcomes.deleted, 1);
5098 assert_eq!(storage_shard_layout(&provider, addr, slot), vec![(u64::MAX, vec![300])]);
5099 }
5100
5101 #[test]
5102 fn test_prune_account_history_batch_seeks_after_stopping_early() {
5103 let temp_dir = TempDir::new().unwrap();
5104 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
5105
5106 let addr1 = Address::repeat_byte(0x01);
5107 let addr2 = Address::repeat_byte(0x02);
5108
5109 let mut batch = provider.batch();
5110 batch
5111 .put::<tables::AccountsHistory>(
5112 ShardedKey::new(addr1, 100),
5113 &BlockNumberList::new_pre_sorted([10, 50, 100]),
5114 )
5115 .unwrap();
5116 batch
5117 .put::<tables::AccountsHistory>(
5118 ShardedKey::new(addr1, u64::MAX),
5119 &BlockNumberList::new_pre_sorted([250, 300]),
5120 )
5121 .unwrap();
5122 batch
5123 .put::<tables::AccountsHistory>(
5124 ShardedKey::new(addr2, u64::MAX),
5125 &BlockNumberList::new_pre_sorted([5, 10, 15]),
5126 )
5127 .unwrap();
5128 batch.commit().unwrap();
5129
5130 let mut batch = provider.batch();
5133 let outcomes = batch.prune_account_history_batch(&[(addr1, 50), (addr2, 10)]).unwrap();
5134 batch.commit().unwrap();
5135
5136 assert_eq!(outcomes.updated, 2);
5137 assert_eq!(
5138 account_shard_layout(&provider, addr1),
5139 vec![(100, vec![100]), (u64::MAX, vec![250, 300])]
5140 );
5141 assert_eq!(account_shard_layout(&provider, addr2), vec![(u64::MAX, vec![15])]);
5142 }
5143
5144 #[test]
5145 fn test_prune_storage_history_batch_seeks_after_stopping_early() {
5146 let temp_dir = TempDir::new().unwrap();
5147 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
5148
5149 let addr = Address::repeat_byte(0x42);
5150 let slot1 = B256::repeat_byte(0x01);
5151 let slot2 = B256::repeat_byte(0x02);
5152 seed_three_storage_shards(&provider, addr, slot1);
5153
5154 let mut batch = provider.batch();
5155 batch
5156 .put::<tables::StoragesHistory>(
5157 StorageShardedKey::last(addr, slot2),
5158 &BlockNumberList::new_pre_sorted([20, 40]),
5159 )
5160 .unwrap();
5161 batch.commit().unwrap();
5162
5163 let mut batch = provider.batch();
5166 let outcomes =
5167 batch.prune_storage_history_batch(&[((addr, slot1), 50), ((addr, slot2), 30)]).unwrap();
5168 batch.commit().unwrap();
5169
5170 assert_eq!(outcomes.updated, 2);
5171 assert_eq!(
5172 storage_shard_layout(&provider, addr, slot1),
5173 vec![(100, vec![100]), (200, vec![150, 200]), (u64::MAX, vec![250, 300])]
5174 );
5175 assert_eq!(storage_shard_layout(&provider, addr, slot2), vec![(u64::MAX, vec![40])]);
5176 }
5177}