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 BlockBasedOptions, Cache, ColumnFamilyDescriptor, CompactionPri, DBCompressionType,
30 DBRawIteratorWithThreadMode, IteratorMode, OptimisticTransactionDB,
31 OptimisticTransactionOptions, Options, 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 }
256
257 options
258 }
259
260 fn default_column_family_options(cache: &Cache) -> Options {
262 let table_options = Self::default_table_options(cache);
264
265 let mut cf_options = Options::default();
266 cf_options.set_block_based_table_factory(&table_options);
267 cf_options.set_level_compaction_dynamic_level_bytes(true);
268 cf_options.set_compression_type(DBCompressionType::Lz4);
270 cf_options.set_bottommost_compression_type(DBCompressionType::Zstd);
271 cf_options.set_bottommost_zstd_max_train_bytes(0, true);
273 cf_options.set_write_buffer_size(DEFAULT_WRITE_BUFFER_SIZE);
274
275 cf_options
276 }
277
278 fn block_access_lists_column_family_options(cache: &Cache) -> Options {
280 let mut cf_options = Self::default_column_family_options(cache);
281 cf_options.set_enable_blob_files(true);
282 cf_options.set_min_blob_size(DEFAULT_BAL_MIN_BLOB_SIZE);
283 cf_options.set_blob_file_size(DEFAULT_BAL_BLOB_FILE_SIZE);
284 cf_options.set_blob_compression_type(DBCompressionType::Lz4);
285 cf_options
286 }
287
288 fn tx_hash_numbers_column_family_options(cache: &Cache) -> Options {
295 let mut table_options = BlockBasedOptions::default();
296 table_options.set_block_size(DEFAULT_BLOCK_SIZE);
297 table_options.set_cache_index_and_filter_blocks(true);
298 table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
299 table_options.set_block_cache(cache);
300 let mut cf_options = Options::default();
304 cf_options.set_block_based_table_factory(&table_options);
305 cf_options.set_level_compaction_dynamic_level_bytes(true);
306 cf_options.set_compression_type(DBCompressionType::None);
309 cf_options.set_bottommost_compression_type(DBCompressionType::None);
310
311 cf_options
312 }
313
314 pub fn with_table<T: Table>(mut self) -> Self {
316 self.column_families.push(T::NAME.to_string());
317 self
318 }
319
320 pub fn with_default_tables(self) -> Self {
327 self.with_table::<tables::TransactionHashNumbers>()
328 .with_table::<tables::AccountsHistory>()
329 .with_table::<tables::StoragesHistory>()
330 }
331
332 pub const fn with_metrics(mut self) -> Self {
334 self.enable_metrics = true;
335 self
336 }
337
338 pub const fn with_statistics(mut self) -> Self {
340 self.enable_statistics = true;
341 self
342 }
343
344 pub const fn with_database_log_level(mut self, log_level: Option<LogLevel>) -> Self {
346 if let Some(level) = log_level {
347 self.log_level = convert_log_level(level);
348 }
349 self
350 }
351
352 pub fn with_block_cache_size(mut self, capacity_bytes: usize) -> Self {
354 self.block_cache = Cache::new_lru_cache(capacity_bytes);
355 self
356 }
357
358 pub const fn with_read_only(mut self, read_only: bool) -> Self {
366 self.read_only = read_only;
367 self
368 }
369
370 pub fn build(self) -> ProviderResult<RocksDBProvider> {
372 let options =
373 Self::default_options(self.log_level, &self.block_cache, self.enable_statistics);
374
375 let mut cf_descriptors: Vec<ColumnFamilyDescriptor> = self
376 .column_families
377 .iter()
378 .map(|name| {
379 let cf_options = if name == tables::TransactionHashNumbers::NAME {
380 Self::tx_hash_numbers_column_family_options(&self.block_cache)
381 } else if name == tables::BlockAccessLists::NAME {
382 Self::block_access_lists_column_family_options(&self.block_cache)
383 } else {
384 Self::default_column_family_options(&self.block_cache)
385 };
386 ColumnFamilyDescriptor::new(name.clone(), cf_options)
387 })
388 .collect();
389
390 if RocksDBProvider::exists(&self.path) {
393 let existing_column_families = DB::list_cf(&options, &self.path).map_err(|e| {
394 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
395 message: e.to_string().into(),
396 code: -1,
397 }))
398 })?;
399 let unknown_column_families: Vec<String> = existing_column_families
400 .into_iter()
401 .filter(|name| {
402 name != DEFAULT_COLUMN_FAMILY_NAME && !self.column_families.contains(name)
403 })
404 .collect();
405 if !unknown_column_families.is_empty() {
406 tracing::debug!(
407 target: "providers::rocksdb",
408 column_families = ?unknown_column_families,
409 "Preserving unknown column families"
410 );
411 cf_descriptors.extend(unknown_column_families.into_iter().map(|name| {
412 ColumnFamilyDescriptor::new(
413 name,
414 Self::default_column_family_options(&self.block_cache),
415 )
416 }));
417 }
418 }
419
420 let metrics = self.enable_metrics.then(RocksDBMetrics::default);
421
422 if self.read_only {
423 let mut options = options;
426 options.set_max_open_files(KEEP_ALL_FILES_OPEN);
427
428 let secondary_path = self
429 .path
430 .parent()
431 .unwrap_or(&self.path)
432 .join(format!("rocksdb-secondary-tmp-{}", std::process::id()));
433 reth_fs_util::create_dir_all(&secondary_path).map_err(ProviderError::other)?;
434
435 let db = DB::open_cf_descriptors_as_secondary(
436 &options,
437 &self.path,
438 &secondary_path,
439 cf_descriptors,
440 )
441 .map_err(|e| {
442 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
443 message: e.to_string().into(),
444 code: -1,
445 }))
446 })?;
447 Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::Secondary {
448 db,
449 metrics,
450 secondary_path,
451 })))
452 } else {
453 let db =
458 OptimisticTransactionDB::open_cf_descriptors(&options, &self.path, cf_descriptors)
459 .map_err(|e| {
460 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
461 message: e.to_string().into(),
462 code: -1,
463 }))
464 })?;
465 Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::ReadWrite { db, metrics })))
466 }
467 }
468}
469
470macro_rules! compress_to_buf_or_ref {
473 ($buf:expr, $value:expr) => {
474 if let Some(value) = $value.uncompressable_ref() {
475 Some(value)
476 } else {
477 $buf.clear();
478 $value.compress_to_buf(&mut $buf);
479 None
480 }
481 };
482}
483
484#[derive(Debug)]
486pub struct RocksDBProvider(Arc<RocksDBProviderInner>);
487
488enum RocksDBProviderInner {
490 ReadWrite {
492 db: OptimisticTransactionDB,
494 metrics: Option<RocksDBMetrics>,
496 },
497 Secondary {
501 db: DB,
503 metrics: Option<RocksDBMetrics>,
505 secondary_path: PathBuf,
507 },
508}
509
510impl RocksDBProviderInner {
511 const fn metrics(&self) -> Option<&RocksDBMetrics> {
513 match self {
514 Self::ReadWrite { metrics, .. } | Self::Secondary { metrics, .. } => metrics.as_ref(),
515 }
516 }
517
518 fn db_rw(&self) -> &OptimisticTransactionDB {
520 match self {
521 Self::ReadWrite { db, .. } => db,
522 Self::Secondary { .. } => {
523 panic!("Cannot perform write operation on secondary RocksDB provider")
524 }
525 }
526 }
527
528 fn cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
530 let cf = match self {
531 Self::ReadWrite { db, .. } => db.cf_handle(T::NAME),
532 Self::Secondary { db, .. } => db.cf_handle(T::NAME),
533 };
534 cf.ok_or_else(|| DatabaseError::Other(format!("Column family '{}' not found", T::NAME)))
535 }
536
537 fn get_cf(
539 &self,
540 cf: &rocksdb::ColumnFamily,
541 key: impl AsRef<[u8]>,
542 ) -> Result<Option<Vec<u8>>, rocksdb::Error> {
543 match self {
544 Self::ReadWrite { db, .. } => db.get_cf(cf, key),
545 Self::Secondary { db, .. } => db.get_cf(cf, key),
546 }
547 }
548
549 fn put_cf(
551 &self,
552 cf: &rocksdb::ColumnFamily,
553 key: impl AsRef<[u8]>,
554 value: impl AsRef<[u8]>,
555 ) -> Result<(), rocksdb::Error> {
556 self.db_rw().put_cf(cf, key, value)
557 }
558
559 fn delete_cf(
561 &self,
562 cf: &rocksdb::ColumnFamily,
563 key: impl AsRef<[u8]>,
564 ) -> Result<(), rocksdb::Error> {
565 self.db_rw().delete_cf(cf, key)
566 }
567
568 fn delete_range_cf<K: AsRef<[u8]>>(
570 &self,
571 cf: &rocksdb::ColumnFamily,
572 from: K,
573 to: K,
574 ) -> Result<(), rocksdb::Error> {
575 self.db_rw().delete_range_cf(cf, from, to)
576 }
577
578 fn iterator_cf(
580 &self,
581 cf: &rocksdb::ColumnFamily,
582 mode: IteratorMode<'_>,
583 ) -> RocksDBIterEnum<'_> {
584 match self {
585 Self::ReadWrite { db, .. } => RocksDBIterEnum::ReadWrite(db.iterator_cf(cf, mode)),
586 Self::Secondary { db, .. } => RocksDBIterEnum::ReadOnly(db.iterator_cf(cf, mode)),
587 }
588 }
589
590 fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
595 match self {
596 Self::ReadWrite { db, .. } => RocksDBRawIterEnum::ReadWrite(db.raw_iterator_cf(cf)),
597 Self::Secondary { db, .. } => RocksDBRawIterEnum::ReadOnly(db.raw_iterator_cf(cf)),
598 }
599 }
600
601 fn snapshot(&self) -> RocksReadSnapshotInner<'_> {
603 match self {
604 Self::ReadWrite { db, .. } => RocksReadSnapshotInner::ReadWrite(db.snapshot()),
605 Self::Secondary { db, .. } => RocksReadSnapshotInner::Secondary(db),
606 }
607 }
608
609 fn path(&self) -> &Path {
611 match self {
612 Self::ReadWrite { db, .. } => db.path(),
613 Self::Secondary { db, .. } => db.path(),
614 }
615 }
616
617 fn wal_size_bytes(&self) -> u64 {
621 let path = self.path();
622
623 match std::fs::read_dir(path) {
624 Ok(entries) => entries
625 .filter_map(|e| e.ok())
626 .filter(|e| e.path().extension().is_some_and(|ext| ext == "log"))
627 .filter_map(|e| e.metadata().ok())
628 .map(|m| m.len())
629 .sum(),
630 Err(_) => 0,
631 }
632 }
633
634 fn table_stats(&self) -> Vec<RocksDBTableStats> {
636 let mut stats = Vec::new();
637
638 macro_rules! collect_stats {
639 ($db:expr) => {
640 for cf_name in ROCKSDB_TABLES {
641 if let Some(cf) = $db.cf_handle(cf_name) {
642 let estimated_num_keys = $db
643 .property_int_value_cf(cf, rocksdb::properties::ESTIMATE_NUM_KEYS)
644 .ok()
645 .flatten()
646 .unwrap_or(0);
647
648 let sst_size = $db
650 .property_int_value_cf(cf, rocksdb::properties::LIVE_SST_FILES_SIZE)
651 .ok()
652 .flatten()
653 .unwrap_or(0);
654
655 let memtable_size = $db
656 .property_int_value_cf(cf, rocksdb::properties::SIZE_ALL_MEM_TABLES)
657 .ok()
658 .flatten()
659 .unwrap_or(0);
660
661 let estimated_size_bytes = sst_size + memtable_size;
662
663 let pending_compaction_bytes = $db
664 .property_int_value_cf(
665 cf,
666 rocksdb::properties::ESTIMATE_PENDING_COMPACTION_BYTES,
667 )
668 .ok()
669 .flatten()
670 .unwrap_or(0);
671
672 stats.push(RocksDBTableStats {
673 sst_size_bytes: sst_size,
674 memtable_size_bytes: memtable_size,
675 name: cf_name.to_string(),
676 estimated_num_keys,
677 estimated_size_bytes,
678 pending_compaction_bytes,
679 });
680 }
681 }
682 };
683 }
684
685 match self {
686 Self::ReadWrite { db, .. } => collect_stats!(db),
687 Self::Secondary { db, .. } => collect_stats!(db),
688 }
689
690 stats
691 }
692
693 fn db_stats(&self) -> RocksDBStats {
695 RocksDBStats { tables: self.table_stats(), wal_size_bytes: self.wal_size_bytes() }
696 }
697}
698
699impl fmt::Debug for RocksDBProviderInner {
700 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
701 match self {
702 Self::ReadWrite { metrics, .. } => f
703 .debug_struct("RocksDBProviderInner::ReadWrite")
704 .field("db", &"<OptimisticTransactionDB>")
705 .field("metrics", metrics)
706 .finish(),
707 Self::Secondary { metrics, .. } => f
708 .debug_struct("RocksDBProviderInner::Secondary")
709 .field("db", &"<DB (secondary)>")
710 .field("metrics", metrics)
711 .finish(),
712 }
713 }
714}
715
716impl Drop for RocksDBProviderInner {
717 fn drop(&mut self) {
718 match self {
719 Self::ReadWrite { db, .. } => {
720 if let Err(e) = db.flush_wal(true) {
723 tracing::warn!(target: "providers::rocksdb", ?e, "Failed to flush WAL on drop");
724 }
725 for cf_name in ROCKSDB_TABLES {
726 if let Some(cf) = db.cf_handle(cf_name) &&
727 let Err(e) = db.flush_cf(&cf)
728 {
729 tracing::warn!(target: "providers::rocksdb", cf = cf_name, ?e, "Failed to flush CF on drop");
730 }
731 }
732 db.cancel_all_background_work(true);
733 }
734 Self::Secondary { db, secondary_path, .. } => {
735 db.cancel_all_background_work(true);
736 let _ = std::fs::remove_dir_all(secondary_path);
737 }
738 }
739 }
740}
741
742impl Clone for RocksDBProvider {
743 fn clone(&self) -> Self {
744 Self(self.0.clone())
745 }
746}
747
748impl DatabaseMetrics for RocksDBProvider {
749 fn gauge_metrics(&self) -> Vec<(&'static str, f64, Vec<Label>)> {
750 let mut metrics = Vec::new();
751
752 for stat in self.table_stats() {
753 metrics.push((
754 "rocksdb.table_size",
755 stat.estimated_size_bytes as f64,
756 vec![Label::new("table", stat.name.clone())],
757 ));
758 metrics.push((
759 "rocksdb.table_entries",
760 stat.estimated_num_keys as f64,
761 vec![Label::new("table", stat.name.clone())],
762 ));
763 metrics.push((
764 "rocksdb.pending_compaction_bytes",
765 stat.pending_compaction_bytes as f64,
766 vec![Label::new("table", stat.name.clone())],
767 ));
768 metrics.push((
769 "rocksdb.sst_size",
770 stat.sst_size_bytes as f64,
771 vec![Label::new("table", stat.name.clone())],
772 ));
773 metrics.push((
774 "rocksdb.memtable_size",
775 stat.memtable_size_bytes as f64,
776 vec![Label::new("table", stat.name)],
777 ));
778 }
779
780 metrics.push(("rocksdb.wal_size", self.wal_size_bytes() as f64, vec![]));
782
783 metrics
784 }
785}
786
787impl RocksDBProvider {
788 pub fn new(path: impl AsRef<Path>) -> ProviderResult<Self> {
790 RocksDBBuilder::new(path).build()
791 }
792
793 pub fn builder(path: impl AsRef<Path>) -> RocksDBBuilder {
795 RocksDBBuilder::new(path)
796 }
797
798 pub fn exists(path: impl AsRef<Path>) -> bool {
803 path.as_ref().join("CURRENT").exists()
804 }
805
806 pub fn is_read_only(&self) -> bool {
808 matches!(self.0.as_ref(), RocksDBProviderInner::Secondary { .. })
809 }
810
811 pub fn try_catch_up_with_primary(&self) -> ProviderResult<()> {
816 match self.0.as_ref() {
817 RocksDBProviderInner::Secondary { db, .. } => {
818 db.try_catch_up_with_primary().map_err(|e| {
819 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
820 message: e.to_string().into(),
821 code: -1,
822 }))
823 })
824 }
825 _ => Ok(()),
826 }
827 }
828
829 pub fn snapshot(&self) -> RocksReadSnapshot<'_> {
833 RocksReadSnapshot { inner: self.0.snapshot(), provider: self }
834 }
835
836 pub fn tx(&self) -> RocksTx<'_> {
844 let write_options = synced_write_options();
845 let txn_options = OptimisticTransactionOptions::default();
846 let inner = self.0.db_rw().transaction_opt(&write_options, &txn_options);
847 RocksTx { inner, provider: self }
848 }
849
850 pub fn batch(&self) -> RocksDBBatch<'_> {
858 RocksDBBatch {
859 provider: self,
860 inner: WriteBatchWithTransaction::<true>::default(),
861 buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
862 auto_commit_threshold: None,
863 }
864 }
865
866 pub fn batch_with_auto_commit(&self) -> RocksDBBatch<'_> {
872 RocksDBBatch {
873 provider: self,
874 inner: WriteBatchWithTransaction::<true>::default(),
875 buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
876 auto_commit_threshold: Some(DEFAULT_AUTO_COMMIT_THRESHOLD),
877 }
878 }
879
880 fn get_cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
882 self.0.cf_handle::<T>()
883 }
884
885 fn execute_with_operation_metric<R>(
887 &self,
888 operation: RocksDBOperation,
889 table: &'static str,
890 f: impl FnOnce(&Self) -> R,
891 ) -> R {
892 let start = self.0.metrics().map(|_| Instant::now());
893 let res = f(self);
894
895 if let (Some(start), Some(metrics)) = (start, self.0.metrics()) {
896 metrics.record_operation(operation, table, start.elapsed());
897 }
898
899 res
900 }
901
902 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
904 self.get_encoded::<T>(&key.encode())
905 }
906
907 pub fn get_encoded<T: Table>(
909 &self,
910 key: &<T::Key as Encode>::Encoded,
911 ) -> ProviderResult<Option<T::Value>> {
912 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
913 let result = this.0.get_cf(this.get_cf_handle::<T>()?, key.as_ref()).map_err(|e| {
914 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
915 message: e.to_string().into(),
916 code: -1,
917 }))
918 })?;
919
920 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
921 })
922 }
923
924 pub fn get_raw<T: Table>(&self, key: T::Key) -> ProviderResult<Option<Vec<u8>>> {
926 let encoded = key.encode();
927 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
928 this.0.get_cf(this.get_cf_handle::<T>()?, encoded.as_ref()).map_err(|e| {
929 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
930 message: e.to_string().into(),
931 code: -1,
932 }))
933 })
934 })
935 }
936
937 pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
942 let encoded_key = key.encode();
943 self.put_encoded::<T>(&encoded_key, value)
944 }
945
946 pub fn put_encoded<T: Table>(
951 &self,
952 key: &<T::Key as Encode>::Encoded,
953 value: &T::Value,
954 ) -> ProviderResult<()> {
955 self.execute_with_operation_metric(RocksDBOperation::Put, T::NAME, |this| {
956 let mut buf = Vec::new();
960 let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
961
962 this.0.put_cf(this.get_cf_handle::<T>()?, key, value_bytes).map_err(|e| {
963 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
964 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
965 operation: DatabaseWriteOperation::PutUpsert,
966 table_name: T::NAME,
967 key: key.as_ref().to_vec(),
968 })))
969 })
970 })
971 }
972
973 pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
978 self.execute_with_operation_metric(RocksDBOperation::Delete, T::NAME, |this| {
979 this.0.delete_cf(this.get_cf_handle::<T>()?, key.encode().as_ref()).map_err(|e| {
980 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
981 message: e.to_string().into(),
982 code: -1,
983 }))
984 })
985 })
986 }
987
988 pub fn clear<T: Table>(&self) -> ProviderResult<()> {
994 let cf = self.get_cf_handle::<T>()?;
995
996 self.0.delete_range_cf(cf, &[] as &[u8], &[0xFF; 256]).map_err(|e| {
997 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
998 message: e.to_string().into(),
999 code: -1,
1000 }))
1001 })?;
1002
1003 Ok(())
1004 }
1005
1006 fn get_boundary<T: Table>(
1008 &self,
1009 mode: IteratorMode<'_>,
1010 ) -> ProviderResult<Option<(T::Key, T::Value)>> {
1011 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
1012 let cf = this.get_cf_handle::<T>()?;
1013 let mut iter = this.0.iterator_cf(cf, mode);
1014
1015 match iter.next() {
1016 Some(Ok((key_bytes, value_bytes))) => {
1017 let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
1018 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1019 let value = T::Value::decompress(&value_bytes)
1020 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1021 Ok(Some((key, value)))
1022 }
1023 Some(Err(e)) => {
1024 Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1025 message: e.to_string().into(),
1026 code: -1,
1027 })))
1028 }
1029 None => Ok(None),
1030 }
1031 })
1032 }
1033
1034 #[inline]
1036 pub fn first<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1037 self.get_boundary::<T>(IteratorMode::Start)
1038 }
1039
1040 #[inline]
1042 pub fn last<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1043 self.get_boundary::<T>(IteratorMode::End)
1044 }
1045
1046 pub fn iter<T: Table>(&self) -> ProviderResult<RocksDBIter<'_, T>> {
1050 let cf = self.get_cf_handle::<T>()?;
1051 let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1052 Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1053 }
1054
1055 pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksDBIter<'_, T>> {
1059 let cf = self.get_cf_handle::<T>()?;
1060 let encoded_key = key.encode();
1061 let iter = self
1062 .0
1063 .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
1064 Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1065 }
1066
1067 pub fn table_stats(&self) -> Vec<RocksDBTableStats> {
1071 self.0.table_stats()
1072 }
1073
1074 pub fn wal_size_bytes(&self) -> u64 {
1080 self.0.wal_size_bytes()
1081 }
1082
1083 pub fn db_stats(&self) -> RocksDBStats {
1087 self.0.db_stats()
1088 }
1089
1090 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(tables = ?tables))]
1101 pub fn flush(&self, tables: &[&'static str]) -> ProviderResult<()> {
1102 let db = self.0.db_rw();
1103
1104 for cf_name in tables {
1105 if let Some(cf) = db.cf_handle(cf_name) {
1106 db.flush_cf(&cf).map_err(|e| {
1107 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1108 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1109 operation: DatabaseWriteOperation::Flush,
1110 table_name: cf_name,
1111 key: Vec::new(),
1112 })))
1113 })?;
1114 }
1115 }
1116
1117 db.flush_wal(true).map_err(|e| {
1118 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1119 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1120 operation: DatabaseWriteOperation::Flush,
1121 table_name: "WAL",
1122 key: Vec::new(),
1123 })))
1124 })?;
1125
1126 Ok(())
1127 }
1128
1129 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1141 pub fn flush_and_compact(&self) -> ProviderResult<()> {
1142 self.flush(ROCKSDB_TABLES)?;
1143
1144 let db = self.0.db_rw();
1145
1146 for cf_name in ROCKSDB_TABLES {
1147 if let Some(cf) = db.cf_handle(cf_name) {
1148 db.compact_range_cf(&cf, None::<&[u8]>, None::<&[u8]>);
1149 }
1150 }
1151
1152 Ok(())
1153 }
1154
1155 pub fn raw_iter<T: Table>(&self) -> ProviderResult<RocksDBRawIter<'_>> {
1159 let cf = self.get_cf_handle::<T>()?;
1160 let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1161 Ok(RocksDBRawIter { inner: iter })
1162 }
1163
1164 pub(crate) fn raw_key_iter_from<T: Table>(
1166 &self,
1167 key: T::Key,
1168 ) -> ProviderResult<RocksDBRawKeyIter<'_>> {
1169 let cf = self.get_cf_handle::<T>()?;
1170 let encoded_key = key.encode();
1171 let mut iter = self.0.raw_iterator_cf(cf);
1172 iter.seek(encoded_key.as_ref());
1173 Ok(RocksDBRawKeyIter { inner: iter })
1174 }
1175
1176 pub fn account_history_shards(
1181 &self,
1182 address: Address,
1183 ) -> ProviderResult<Vec<(ShardedKey<Address>, BlockNumberList)>> {
1184 let cf = self.get_cf_handle::<tables::AccountsHistory>()?;
1186
1187 let start_key = ShardedKey::new(address, 0u64);
1190 let start_bytes = start_key.encode();
1191
1192 let iter = self
1194 .0
1195 .iterator_cf(cf, IteratorMode::From(start_bytes.as_ref(), rocksdb::Direction::Forward));
1196
1197 let mut result = Vec::new();
1198 for item in iter {
1199 match item {
1200 Ok((key_bytes, value_bytes)) => {
1201 let key = ShardedKey::<Address>::decode(&key_bytes)
1203 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1204
1205 if key.key != address {
1207 break;
1208 }
1209
1210 let value = BlockNumberList::decompress(&value_bytes)
1212 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1213
1214 result.push((key, value));
1215 }
1216 Err(e) => {
1217 return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1218 message: e.to_string().into(),
1219 code: -1,
1220 })));
1221 }
1222 }
1223 }
1224
1225 Ok(result)
1226 }
1227
1228 pub fn storage_history_shards(
1233 &self,
1234 address: Address,
1235 storage_key: B256,
1236 ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1237 let cf = self.get_cf_handle::<tables::StoragesHistory>()?;
1238
1239 let start_key = StorageShardedKey::new(address, storage_key, 0u64);
1240 let start_bytes = start_key.encode();
1241
1242 let iter = self
1243 .0
1244 .iterator_cf(cf, IteratorMode::From(start_bytes.as_ref(), rocksdb::Direction::Forward));
1245
1246 let mut result = Vec::new();
1247 for item in iter {
1248 match item {
1249 Ok((key_bytes, value_bytes)) => {
1250 let key = StorageShardedKey::decode(&key_bytes)
1251 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1252
1253 if key.address != address || key.sharded_key.key != storage_key {
1254 break;
1255 }
1256
1257 let value = BlockNumberList::decompress(&value_bytes)
1258 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1259
1260 result.push((key, value));
1261 }
1262 Err(e) => {
1263 return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1264 message: e.to_string().into(),
1265 code: -1,
1266 })));
1267 }
1268 }
1269 }
1270
1271 Ok(result)
1272 }
1273
1274 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1282 pub fn unwind_account_history_indices(
1283 &self,
1284 last_indices: &[(Address, BlockNumber)],
1285 ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1286 let mut address_min_block: AddressMap<BlockNumber> =
1287 AddressMap::with_capacity_and_hasher(last_indices.len(), Default::default());
1288 for &(address, block_number) in last_indices {
1289 address_min_block
1290 .entry(address)
1291 .and_modify(|min| *min = (*min).min(block_number))
1292 .or_insert(block_number);
1293 }
1294
1295 let mut batch = self.batch();
1296 for (address, min_block) in address_min_block {
1297 match min_block.checked_sub(1) {
1298 Some(keep_to) => batch.unwind_account_history_to(address, keep_to)?,
1299 None => batch.clear_account_history(address)?,
1300 }
1301 }
1302
1303 Ok(batch.into_inner())
1304 }
1305
1306 pub fn unwind_storage_history_indices(
1314 &self,
1315 storage_changesets: &[(Address, B256, BlockNumber)],
1316 ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1317 let mut key_min_block: HashMap<(Address, B256), BlockNumber> =
1318 HashMap::with_capacity_and_hasher(storage_changesets.len(), Default::default());
1319 for &(address, storage_key, block_number) in storage_changesets {
1320 key_min_block
1321 .entry((address, storage_key))
1322 .and_modify(|min| *min = (*min).min(block_number))
1323 .or_insert(block_number);
1324 }
1325
1326 let mut batch = self.batch();
1327 for ((address, storage_key), min_block) in key_min_block {
1328 match min_block.checked_sub(1) {
1329 Some(keep_to) => batch.unwind_storage_history_to(address, storage_key, keep_to)?,
1330 None => batch.clear_storage_history(address, storage_key)?,
1331 }
1332 }
1333
1334 Ok(batch.into_inner())
1335 }
1336
1337 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1339 pub fn write_batch<F>(&self, f: F) -> ProviderResult<()>
1340 where
1341 F: FnOnce(&mut RocksDBBatch<'_>) -> ProviderResult<()>,
1342 {
1343 self.execute_with_operation_metric(RocksDBOperation::BatchWrite, "Batch", |this| {
1344 let mut batch_handle = this.batch();
1345 f(&mut batch_handle)?;
1346 batch_handle.commit()
1347 })
1348 }
1349
1350 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = batch.len(), batch_size = batch.size_in_bytes()))]
1358 pub fn commit_batch(&self, batch: WriteBatchWithTransaction<true>) -> ProviderResult<()> {
1359 self.0.db_rw().write_opt(batch, &synced_write_options()).map_err(|e| {
1360 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1361 message: e.to_string().into(),
1362 code: -1,
1363 }))
1364 })
1365 }
1366
1367 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(num_blocks = blocks.len(), first_block = ctx.first_block_number))]
1373 pub(crate) fn write_blocks_data<N: reth_node_types::NodePrimitives>(
1374 &self,
1375 blocks: &[ExecutedBlock<N>],
1376 tx_nums: &[TxNumber],
1377 ctx: RocksDBWriteCtx,
1378 runtime: &reth_tasks::Runtime,
1379 ) -> ProviderResult<()> {
1380 if !ctx.storage_settings.storage_v2 {
1381 return Ok(());
1382 }
1383
1384 let mut r_tx_hash = None;
1385 let mut r_account_history = None;
1386 let mut r_storage_history = None;
1387
1388 let write_tx_hash =
1389 ctx.storage_settings.storage_v2 && ctx.prune_tx_lookup.is_none_or(|m| !m.is_full());
1390 let write_account_history = ctx.storage_settings.storage_v2;
1391 let write_storage_history = ctx.storage_settings.storage_v2;
1392
1393 let span = tracing::Span::current();
1396 runtime.storage_pool().in_place_scope(|s| {
1397 if write_tx_hash {
1398 s.spawn(|_| {
1399 let _guard = span.enter();
1400 r_tx_hash = Some(self.write_tx_hash_numbers(blocks, tx_nums, &ctx));
1401 });
1402 }
1403
1404 if write_account_history {
1405 s.spawn(|_| {
1406 let _guard = span.enter();
1407 r_account_history = Some(self.write_account_history(blocks, &ctx));
1408 });
1409 }
1410
1411 if write_storage_history {
1412 s.spawn(|_| {
1413 let _guard = span.enter();
1414 r_storage_history = Some(self.write_storage_history(blocks, &ctx));
1415 });
1416 }
1417 });
1418
1419 if write_tx_hash {
1420 r_tx_hash.ok_or_else(|| {
1421 ProviderError::Database(DatabaseError::Other(
1422 "rocksdb tx-hash write thread panicked".into(),
1423 ))
1424 })??;
1425 }
1426 if write_account_history {
1427 r_account_history.ok_or_else(|| {
1428 ProviderError::Database(DatabaseError::Other(
1429 "rocksdb account-history write thread panicked".into(),
1430 ))
1431 })??;
1432 }
1433 if write_storage_history {
1434 r_storage_history.ok_or_else(|| {
1435 ProviderError::Database(DatabaseError::Other(
1436 "rocksdb storage-history write thread panicked".into(),
1437 ))
1438 })??;
1439 }
1440
1441 Ok(())
1442 }
1443
1444 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1446 fn write_tx_hash_numbers<N: reth_node_types::NodePrimitives>(
1447 &self,
1448 blocks: &[ExecutedBlock<N>],
1449 tx_nums: &[TxNumber],
1450 ctx: &RocksDBWriteCtx,
1451 ) -> ProviderResult<()> {
1452 let mut batch = self.batch();
1453 for (block, &first_tx_num) in blocks.iter().zip(tx_nums) {
1454 let body = block.recovered_block().body();
1455 for (tx_num, transaction) in (first_tx_num..).zip(body.transactions_iter()) {
1456 batch.put::<tables::TransactionHashNumbers>(*transaction.tx_hash(), &tx_num)?;
1457 }
1458 }
1459 ctx.pending_batches.lock().push(batch.into_inner());
1460 Ok(())
1461 }
1462
1463 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1467 fn write_account_history<N: reth_node_types::NodePrimitives>(
1468 &self,
1469 blocks: &[ExecutedBlock<N>],
1470 ctx: &RocksDBWriteCtx,
1471 ) -> ProviderResult<()> {
1472 let mut batch = self.batch();
1473 let mut account_history: BTreeMap<Address, Vec<u64>> = BTreeMap::new();
1474
1475 for (block_idx, block) in blocks.iter().enumerate() {
1476 let block_number = ctx.first_block_number + block_idx as u64;
1477 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1478
1479 for account_block_reverts in reverts.accounts {
1482 for (address, _) in account_block_reverts {
1483 account_history.entry(address).or_default().push(block_number);
1484 }
1485 }
1486 }
1487
1488 for (address, indices) in account_history {
1490 batch.append_account_history_shard(address, indices)?;
1491 }
1492 ctx.pending_batches.lock().push(batch.into_inner());
1493 Ok(())
1494 }
1495
1496 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1500 fn write_storage_history<N: reth_node_types::NodePrimitives>(
1501 &self,
1502 blocks: &[ExecutedBlock<N>],
1503 ctx: &RocksDBWriteCtx,
1504 ) -> ProviderResult<()> {
1505 let mut storage_history: BTreeMap<(Address, B256), Vec<u64>> = BTreeMap::new();
1506
1507 for (block_idx, block) in blocks.iter().enumerate() {
1508 let block_number = ctx.first_block_number + block_idx as u64;
1509 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1510
1511 for storage_block_reverts in reverts.storage {
1514 for revert in storage_block_reverts {
1515 for (slot, _) in revert.storage_revert {
1516 let plain_key = B256::new(slot.to_be_bytes());
1517 storage_history
1518 .entry((revert.address, plain_key))
1519 .or_default()
1520 .push(block_number);
1521 }
1522 }
1523 }
1524 }
1525
1526 let shard_puts = storage_history
1527 .into_par_iter()
1528 .map(|((address, slot), indices)| {
1529 self.storage_history_shards_to_put(address, slot, indices)
1530 })
1531 .collect::<ProviderResult<Vec<_>>>()?;
1532
1533 let mut batch = self.batch();
1534 for shards in shard_puts {
1535 for (key, shard) in shards {
1536 batch.put::<tables::StoragesHistory>(key, &shard)?;
1537 }
1538 }
1539 ctx.pending_batches.lock().push(batch.into_inner());
1540 Ok(())
1541 }
1542
1543 fn storage_history_shards_to_put(
1546 &self,
1547 address: Address,
1548 storage_key: B256,
1549 indices: Vec<u64>,
1550 ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1551 if indices.is_empty() {
1552 return Ok(Vec::new());
1553 }
1554
1555 debug_assert!(
1556 indices.windows(2).all(|w| w[0] < w[1]),
1557 "indices must be strictly increasing: {:?}",
1558 indices
1559 );
1560
1561 let last_key = StorageShardedKey::last(address, storage_key);
1562 let last_shard_opt = self.get::<tables::StoragesHistory>(last_key.clone())?;
1563 let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
1564
1565 last_shard.append(indices).map_err(ProviderError::other)?;
1566
1567 if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
1568 return Ok(vec![(last_key, last_shard)]);
1569 }
1570
1571 let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
1572 let mut chunks_peekable = chunks.into_iter().peekable();
1573 let mut shards = Vec::new();
1574
1575 while let Some(chunk) = chunks_peekable.next() {
1576 let shard = BlockNumberList::new_pre_sorted(chunk);
1577 let highest_block_number = if chunks_peekable.peek().is_some() {
1578 shard.iter().next_back().expect("`chunks` does not return empty list")
1579 } else {
1580 u64::MAX
1581 };
1582
1583 shards
1584 .push((StorageShardedKey::new(address, storage_key, highest_block_number), shard));
1585 }
1586
1587 Ok(shards)
1588 }
1589}
1590
1591pub struct RocksReadSnapshot<'db> {
1599 inner: RocksReadSnapshotInner<'db>,
1600 provider: &'db RocksDBProvider,
1601}
1602
1603enum RocksReadSnapshotInner<'db> {
1605 ReadWrite(SnapshotWithThreadMode<'db, OptimisticTransactionDB>),
1607 Secondary(&'db DB),
1609}
1610
1611impl<'db> RocksReadSnapshotInner<'db> {
1612 fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
1614 match self {
1615 Self::ReadWrite(snap) => RocksDBRawIterEnum::ReadWrite(snap.raw_iterator_cf(cf)),
1616 Self::Secondary(db) => RocksDBRawIterEnum::ReadOnly(db.raw_iterator_cf(cf)),
1617 }
1618 }
1619}
1620
1621impl fmt::Debug for RocksReadSnapshot<'_> {
1622 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1623 f.debug_struct("RocksReadSnapshot")
1624 .field("provider", &self.provider)
1625 .finish_non_exhaustive()
1626 }
1627}
1628
1629impl<'db> RocksReadSnapshot<'db> {
1630 fn cf_handle<T: Table>(&self) -> Result<&'db rocksdb::ColumnFamily, DatabaseError> {
1632 self.provider.get_cf_handle::<T>()
1633 }
1634
1635 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
1637 let encoded_key = key.encode();
1638 let cf = self.cf_handle::<T>()?;
1639 let result = match &self.inner {
1640 RocksReadSnapshotInner::ReadWrite(snap) => snap.get_cf(cf, encoded_key.as_ref()),
1641 RocksReadSnapshotInner::Secondary(db) => db.get_cf(cf, encoded_key.as_ref()),
1642 }
1643 .map_err(|e| {
1644 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1645 message: e.to_string().into(),
1646 code: -1,
1647 }))
1648 })?;
1649
1650 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
1651 }
1652
1653 pub fn account_history_info(
1658 &self,
1659 address: Address,
1660 block_number: BlockNumber,
1661 lowest_available_block_number: Option<BlockNumber>,
1662 visible_tip: BlockNumber,
1663 ) -> ProviderResult<HistoryInfo> {
1664 let key = ShardedKey::new(address, block_number);
1665 self.history_info::<tables::AccountsHistory>(
1666 key.encode().as_ref(),
1667 block_number,
1668 lowest_available_block_number,
1669 visible_tip,
1670 |key_bytes| Ok(<ShardedKey<Address> as Decode>::decode(key_bytes)?.key == address),
1671 |prev_bytes| {
1672 <ShardedKey<Address> as Decode>::decode(prev_bytes)
1673 .map(|k| k.key == address)
1674 .unwrap_or(false)
1675 },
1676 )
1677 }
1678
1679 pub fn storage_history_info(
1684 &self,
1685 address: Address,
1686 storage_key: B256,
1687 block_number: BlockNumber,
1688 lowest_available_block_number: Option<BlockNumber>,
1689 visible_tip: BlockNumber,
1690 ) -> ProviderResult<HistoryInfo> {
1691 let key = StorageShardedKey::new(address, storage_key, block_number);
1692 self.history_info::<tables::StoragesHistory>(
1693 key.encode().as_ref(),
1694 block_number,
1695 lowest_available_block_number,
1696 visible_tip,
1697 |key_bytes| {
1698 let k = <StorageShardedKey as Decode>::decode(key_bytes)?;
1699 Ok(k.address == address && k.sharded_key.key == storage_key)
1700 },
1701 |prev_bytes| {
1702 <StorageShardedKey as Decode>::decode(prev_bytes)
1703 .map(|k| k.address == address && k.sharded_key.key == storage_key)
1704 .unwrap_or(false)
1705 },
1706 )
1707 }
1708
1709 fn history_info<T>(
1715 &self,
1716 encoded_key: &[u8],
1717 block_number: BlockNumber,
1718 lowest_available_block_number: Option<BlockNumber>,
1719 visible_tip: BlockNumber,
1720 key_matches: impl FnOnce(&[u8]) -> Result<bool, reth_db_api::DatabaseError>,
1721 prev_key_matches: impl Fn(&[u8]) -> bool,
1722 ) -> ProviderResult<HistoryInfo>
1723 where
1724 T: Table<Value = BlockNumberList>,
1725 {
1726 let is_maybe_pruned = lowest_available_block_number.is_some();
1727 let fallback = || {
1728 Ok(if is_maybe_pruned {
1729 HistoryInfo::MaybeInPlainState
1730 } else {
1731 HistoryInfo::NotYetWritten
1732 })
1733 };
1734
1735 let cf = self.cf_handle::<T>()?;
1736 let mut iter = self.inner.raw_iterator_cf(cf);
1737
1738 iter.seek(encoded_key);
1739 iter.status().map_err(|e| {
1740 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1741 message: e.to_string().into(),
1742 code: -1,
1743 }))
1744 })?;
1745
1746 if !iter.valid() {
1747 return fallback();
1748 }
1749
1750 let Some(key_bytes) = iter.key() else {
1751 return fallback();
1752 };
1753 if !key_matches(key_bytes)? {
1754 return fallback();
1755 }
1756
1757 let Some(value_bytes) = iter.value() else {
1758 return fallback();
1759 };
1760 let chunk = BlockNumberList::decompress(value_bytes)?;
1761
1762 let (rank, found_block) = compute_history_rank(&chunk, block_number);
1763 let found_block = found_block.filter(|block| *block <= visible_tip);
1765
1766 let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) {
1767 iter.prev();
1768 iter.status().map_err(|e| {
1769 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1770 message: e.to_string().into(),
1771 code: -1,
1772 }))
1773 })?;
1774 let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches);
1775
1776 if found_block.is_none() && !has_prev {
1780 return fallback()
1781 }
1782
1783 !has_prev
1784 } else {
1785 false
1786 };
1787
1788 Ok(HistoryInfo::from_lookup(
1789 found_block,
1790 is_before_first_write,
1791 lowest_available_block_number,
1792 ))
1793 }
1794}
1795
1796#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1798pub enum PruneShardOutcome {
1799 Deleted,
1801 Updated,
1803 Unchanged,
1805}
1806
1807#[derive(Debug, Default, Clone, Copy)]
1809pub struct PrunedIndices {
1810 pub deleted: usize,
1812 pub updated: usize,
1814 pub unchanged: usize,
1816}
1817
1818#[must_use = "batch must be committed"]
1828pub struct RocksDBBatch<'a> {
1829 provider: &'a RocksDBProvider,
1830 inner: WriteBatchWithTransaction<true>,
1831 buf: Vec<u8>,
1832 auto_commit_threshold: Option<usize>,
1834}
1835
1836impl fmt::Debug for RocksDBBatch<'_> {
1837 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1838 f.debug_struct("RocksDBBatch")
1839 .field("provider", &self.provider)
1840 .field("batch", &"<WriteBatchWithTransaction>")
1841 .field("length", &self.inner.len())
1843 .field("size_in_bytes", &self.inner.size_in_bytes())
1846 .finish()
1847 }
1848}
1849
1850impl<'a> RocksDBBatch<'a> {
1851 pub fn put<T: Table>(&mut self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
1855 let encoded_key = key.encode();
1856 self.put_encoded::<T>(&encoded_key, value)
1857 }
1858
1859 pub fn put_encoded<T: Table>(
1863 &mut self,
1864 key: &<T::Key as Encode>::Encoded,
1865 value: &T::Value,
1866 ) -> ProviderResult<()> {
1867 let value_bytes = compress_to_buf_or_ref!(self.buf, value).unwrap_or(&self.buf);
1868 self.inner.put_cf(self.provider.get_cf_handle::<T>()?, key, value_bytes);
1869 self.maybe_auto_commit()?;
1870 Ok(())
1871 }
1872
1873 pub fn delete<T: Table>(&mut self, key: T::Key) -> ProviderResult<()> {
1877 self.inner.delete_cf(self.provider.get_cf_handle::<T>()?, key.encode().as_ref());
1878 self.maybe_auto_commit()?;
1879 Ok(())
1880 }
1881
1882 fn maybe_auto_commit(&mut self) -> ProviderResult<()> {
1887 if let Some(threshold) = self.auto_commit_threshold &&
1888 self.inner.size_in_bytes() >= threshold
1889 {
1890 tracing::debug!(
1891 target: "providers::rocksdb",
1892 batch_size = self.inner.size_in_bytes(),
1893 threshold,
1894 "Auto-committing RocksDB batch"
1895 );
1896 let old_batch = std::mem::take(&mut self.inner);
1897 self.provider.0.db_rw().write_opt(old_batch, &synced_write_options()).map_err(|e| {
1898 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1899 message: e.to_string().into(),
1900 code: -1,
1901 }))
1902 })?;
1903 }
1904 Ok(())
1905 }
1906
1907 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = self.inner.len(), batch_size = self.inner.size_in_bytes()))]
1914 pub fn commit(self) -> ProviderResult<()> {
1915 self.provider.0.db_rw().write_opt(self.inner, &synced_write_options()).map_err(|e| {
1916 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1917 message: e.to_string().into(),
1918 code: -1,
1919 }))
1920 })
1921 }
1922
1923 pub fn len(&self) -> usize {
1925 self.inner.len()
1926 }
1927
1928 pub fn is_empty(&self) -> bool {
1930 self.inner.is_empty()
1931 }
1932
1933 pub fn size_in_bytes(&self) -> usize {
1935 self.inner.size_in_bytes()
1936 }
1937
1938 pub const fn provider(&self) -> &RocksDBProvider {
1940 self.provider
1941 }
1942
1943 pub fn into_inner(self) -> WriteBatchWithTransaction<true> {
1947 self.inner
1948 }
1949
1950 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
1955 self.provider.get::<T>(key)
1956 }
1957
1958 pub fn append_account_history_shard(
1970 &mut self,
1971 address: Address,
1972 indices: impl IntoIterator<Item = u64>,
1973 ) -> ProviderResult<()> {
1974 let indices: Vec<u64> = indices.into_iter().collect();
1975
1976 if indices.is_empty() {
1977 return Ok(());
1978 }
1979
1980 debug_assert!(
1981 indices.windows(2).all(|w| w[0] < w[1]),
1982 "indices must be strictly increasing: {:?}",
1983 indices
1984 );
1985
1986 let last_key = ShardedKey::new(address, u64::MAX);
1987 let last_shard_opt = self.provider.get::<tables::AccountsHistory>(last_key.clone())?;
1988 let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
1989
1990 last_shard.append(indices).map_err(ProviderError::other)?;
1991
1992 if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
1994 self.put::<tables::AccountsHistory>(last_key, &last_shard)?;
1995 return Ok(());
1996 }
1997
1998 let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
2000 let mut chunks_peekable = chunks.into_iter().peekable();
2001
2002 while let Some(chunk) = chunks_peekable.next() {
2003 let shard = BlockNumberList::new_pre_sorted(chunk);
2004 let highest_block_number = if chunks_peekable.peek().is_some() {
2005 shard.iter().next_back().expect("`chunks` does not return empty list")
2006 } else {
2007 u64::MAX
2008 };
2009
2010 self.put::<tables::AccountsHistory>(
2011 ShardedKey::new(address, highest_block_number),
2012 &shard,
2013 )?;
2014 }
2015
2016 Ok(())
2017 }
2018
2019 pub fn append_storage_history_shard(
2031 &mut self,
2032 address: Address,
2033 storage_key: B256,
2034 indices: impl IntoIterator<Item = u64>,
2035 ) -> ProviderResult<()> {
2036 let indices: Vec<u64> = indices.into_iter().collect();
2037
2038 for (key, shard) in
2039 self.provider.storage_history_shards_to_put(address, storage_key, indices)?
2040 {
2041 self.put::<tables::StoragesHistory>(key, &shard)?;
2042 }
2043
2044 Ok(())
2045 }
2046
2047 pub fn unwind_account_history_to(
2054 &mut self,
2055 address: Address,
2056 keep_to: BlockNumber,
2057 ) -> ProviderResult<()> {
2058 let shards = self.provider.account_history_shards(address)?;
2059 if shards.is_empty() {
2060 return Ok(());
2061 }
2062
2063 let boundary_idx = shards.iter().position(|(key, _)| {
2066 key.highest_block_number == u64::MAX || key.highest_block_number > keep_to
2067 });
2068
2069 let Some(boundary_idx) = boundary_idx else {
2071 let (last_key, last_value) = shards.last().expect("shards is non-empty");
2072 if last_key.highest_block_number != u64::MAX {
2073 self.delete::<tables::AccountsHistory>(last_key.clone())?;
2074 self.put::<tables::AccountsHistory>(
2075 ShardedKey::new(address, u64::MAX),
2076 last_value,
2077 )?;
2078 }
2079 return Ok(());
2080 };
2081
2082 for (key, _) in shards.iter().skip(boundary_idx + 1) {
2084 self.delete::<tables::AccountsHistory>(key.clone())?;
2085 }
2086
2087 let (boundary_key, boundary_list) = &shards[boundary_idx];
2089
2090 self.delete::<tables::AccountsHistory>(boundary_key.clone())?;
2092
2093 let new_last =
2095 BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2096
2097 if new_last.is_empty() {
2098 if boundary_idx == 0 {
2101 return Ok(());
2103 }
2104
2105 let (prev_key, prev_value) = &shards[boundary_idx - 1];
2106 if prev_key.highest_block_number != u64::MAX {
2107 self.delete::<tables::AccountsHistory>(prev_key.clone())?;
2108 self.put::<tables::AccountsHistory>(
2109 ShardedKey::new(address, u64::MAX),
2110 prev_value,
2111 )?;
2112 }
2113 return Ok(());
2114 }
2115
2116 self.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &new_last)?;
2117
2118 Ok(())
2119 }
2120
2121 #[expect(clippy::too_many_arguments)]
2130 fn prune_history_shards_inner<K>(
2131 &mut self,
2132 shards: Vec<(K, BlockNumberList)>,
2133 shards_complete: bool,
2134 to_block: BlockNumber,
2135 get_highest: impl Fn(&K) -> u64,
2136 is_sentinel: impl Fn(&K) -> bool,
2137 delete_shard: impl Fn(&mut Self, K) -> ProviderResult<()>,
2138 put_shard: impl Fn(&mut Self, K, &BlockNumberList) -> ProviderResult<()>,
2139 create_sentinel: impl Fn() -> K,
2140 ) -> ProviderResult<PruneShardOutcome>
2141 where
2142 K: Clone,
2143 {
2144 if shards.is_empty() {
2145 return Ok(PruneShardOutcome::Unchanged);
2146 }
2147
2148 let mut deleted = false;
2149 let mut updated = false;
2150 let mut last_remaining: Option<(K, BlockNumberList)> = None;
2151
2152 for (key, mut block_list) in shards {
2153 if !is_sentinel(&key) && get_highest(&key) <= to_block {
2154 delete_shard(self, key)?;
2155 deleted = true;
2156 } else {
2157 let removed = block_list.remove_range(0..=to_block);
2158
2159 if block_list.is_empty() {
2160 delete_shard(self, key)?;
2161 deleted = true;
2162 } else if removed > 0 {
2163 put_shard(self, key.clone(), &block_list)?;
2164 last_remaining = Some((key, block_list));
2165 updated = true;
2166 } else {
2167 last_remaining = Some((key, block_list));
2168 }
2169 }
2170 }
2171
2172 if shards_complete &&
2173 let Some((last_key, last_value)) = last_remaining &&
2174 !is_sentinel(&last_key)
2175 {
2176 delete_shard(self, last_key)?;
2177 put_shard(self, create_sentinel(), &last_value)?;
2178 updated = true;
2179 }
2180
2181 if deleted {
2182 Ok(PruneShardOutcome::Deleted)
2183 } else if updated {
2184 Ok(PruneShardOutcome::Updated)
2185 } else {
2186 Ok(PruneShardOutcome::Unchanged)
2187 }
2188 }
2189
2190 pub fn prune_account_history_to(
2195 &mut self,
2196 address: Address,
2197 to_block: BlockNumber,
2198 ) -> ProviderResult<PruneShardOutcome> {
2199 let shards = self.provider.account_history_shards(address)?;
2200 self.prune_history_shards_inner(
2201 shards,
2202 true,
2203 to_block,
2204 |key| key.highest_block_number,
2205 |key| key.highest_block_number == u64::MAX,
2206 |batch, key| batch.delete::<tables::AccountsHistory>(key),
2207 |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2208 || ShardedKey::new(address, u64::MAX),
2209 )
2210 }
2211
2212 pub fn prune_account_history_batch(
2221 &mut self,
2222 targets: &[(Address, BlockNumber)],
2223 ) -> ProviderResult<PrunedIndices> {
2224 if targets.is_empty() {
2225 return Ok(PrunedIndices::default());
2226 }
2227
2228 debug_assert!(
2229 targets.windows(2).all(|w| w[0].0 < w[1].0),
2230 "prune_account_history_batch: targets must be sorted and unique"
2231 );
2232
2233 const PREFIX_LEN: usize = 20;
2236
2237 let cf = self.provider.get_cf_handle::<tables::AccountsHistory>()?;
2238 let mut iter = self.provider.0.raw_iterator_cf(cf);
2239 let mut outcomes = PrunedIndices::default();
2240
2241 for (address, to_block) in targets {
2242 let start_key = ShardedKey::new(*address, 0u64).encode();
2244 let target_prefix = &start_key[..PREFIX_LEN];
2245
2246 let needs_seek = if iter.valid() {
2254 if let Some(current_key) = iter.key() {
2255 current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2259 } else {
2260 true
2261 }
2262 } else {
2263 true
2264 };
2265
2266 if needs_seek {
2267 iter.seek(start_key);
2268 iter.status().map_err(|e| {
2269 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2270 message: e.to_string().into(),
2271 code: -1,
2272 }))
2273 })?;
2274 }
2275
2276 let mut shards = Vec::new();
2279 let mut shards_complete = true;
2280 while iter.valid() {
2281 let Some(key_bytes) = iter.key() else { break };
2282
2283 let current_prefix = key_bytes.get(..PREFIX_LEN);
2285 if current_prefix != Some(target_prefix) {
2286 break;
2287 }
2288
2289 let key = ShardedKey::<Address>::decode(key_bytes)
2291 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2292
2293 let Some(value_bytes) = iter.value() else { break };
2294 let value = BlockNumberList::decompress(value_bytes)
2295 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2296
2297 let highest = key.highest_block_number;
2298 shards.push((key, value));
2299
2300 iter.next();
2301
2302 if highest > *to_block {
2307 shards_complete = iter.key().and_then(|next_key| next_key.get(..PREFIX_LEN)) !=
2308 Some(target_prefix);
2309 break;
2310 }
2311 }
2312
2313 if !iter.valid() {
2316 iter.status().map_err(|e| {
2317 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2318 message: e.to_string().into(),
2319 code: -1,
2320 }))
2321 })?;
2322 }
2323
2324 match self.prune_history_shards_inner(
2325 shards,
2326 shards_complete,
2327 *to_block,
2328 |key| key.highest_block_number,
2329 |key| key.highest_block_number == u64::MAX,
2330 |batch, key| batch.delete::<tables::AccountsHistory>(key),
2331 |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2332 || ShardedKey::new(*address, u64::MAX),
2333 )? {
2334 PruneShardOutcome::Deleted => outcomes.deleted += 1,
2335 PruneShardOutcome::Updated => outcomes.updated += 1,
2336 PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2337 }
2338 }
2339
2340 Ok(outcomes)
2341 }
2342
2343 pub fn prune_storage_history_to(
2349 &mut self,
2350 address: Address,
2351 storage_key: B256,
2352 to_block: BlockNumber,
2353 ) -> ProviderResult<PruneShardOutcome> {
2354 let shards = self.provider.storage_history_shards(address, storage_key)?;
2355 self.prune_history_shards_inner(
2356 shards,
2357 true,
2358 to_block,
2359 |key| key.sharded_key.highest_block_number,
2360 |key| key.sharded_key.highest_block_number == u64::MAX,
2361 |batch, key| batch.delete::<tables::StoragesHistory>(key),
2362 |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2363 || StorageShardedKey::last(address, storage_key),
2364 )
2365 }
2366
2367 pub fn prune_storage_history_batch(
2377 &mut self,
2378 targets: &[((Address, B256), BlockNumber)],
2379 ) -> ProviderResult<PrunedIndices> {
2380 if targets.is_empty() {
2381 return Ok(PrunedIndices::default());
2382 }
2383
2384 debug_assert!(
2385 targets.windows(2).all(|w| w[0].0 < w[1].0),
2386 "prune_storage_history_batch: targets must be sorted and unique"
2387 );
2388
2389 const PREFIX_LEN: usize = 52;
2392
2393 let cf = self.provider.get_cf_handle::<tables::StoragesHistory>()?;
2394 let mut iter = self.provider.0.raw_iterator_cf(cf);
2395 let mut outcomes = PrunedIndices::default();
2396
2397 for ((address, storage_key), to_block) in targets {
2398 let start_key = StorageShardedKey::new(*address, *storage_key, 0u64).encode();
2400 let target_prefix = &start_key[..PREFIX_LEN];
2401
2402 let needs_seek = if iter.valid() {
2410 if let Some(current_key) = iter.key() {
2411 current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2415 } else {
2416 true
2417 }
2418 } else {
2419 true
2420 };
2421
2422 if needs_seek {
2423 iter.seek(start_key);
2424 iter.status().map_err(|e| {
2425 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2426 message: e.to_string().into(),
2427 code: -1,
2428 }))
2429 })?;
2430 }
2431
2432 let mut shards = Vec::new();
2435 let mut shards_complete = true;
2436 while iter.valid() {
2437 let Some(key_bytes) = iter.key() else { break };
2438
2439 let current_prefix = key_bytes.get(..PREFIX_LEN);
2441 if current_prefix != Some(target_prefix) {
2442 break;
2443 }
2444
2445 let key = StorageShardedKey::decode(key_bytes)
2447 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2448
2449 let Some(value_bytes) = iter.value() else { break };
2450 let value = BlockNumberList::decompress(value_bytes)
2451 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2452
2453 let highest = key.sharded_key.highest_block_number;
2454 shards.push((key, value));
2455
2456 iter.next();
2457
2458 if highest > *to_block {
2463 shards_complete = iter.key().and_then(|next_key| next_key.get(..PREFIX_LEN)) !=
2464 Some(target_prefix);
2465 break;
2466 }
2467 }
2468
2469 if !iter.valid() {
2472 iter.status().map_err(|e| {
2473 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2474 message: e.to_string().into(),
2475 code: -1,
2476 }))
2477 })?;
2478 }
2479
2480 match self.prune_history_shards_inner(
2482 shards,
2483 shards_complete,
2484 *to_block,
2485 |key| key.sharded_key.highest_block_number,
2486 |key| key.sharded_key.highest_block_number == u64::MAX,
2487 |batch, key| batch.delete::<tables::StoragesHistory>(key),
2488 |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2489 || StorageShardedKey::last(*address, *storage_key),
2490 )? {
2491 PruneShardOutcome::Deleted => outcomes.deleted += 1,
2492 PruneShardOutcome::Updated => outcomes.updated += 1,
2493 PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2494 }
2495 }
2496
2497 Ok(outcomes)
2498 }
2499
2500 pub fn unwind_storage_history_to(
2509 &mut self,
2510 address: Address,
2511 storage_key: B256,
2512 keep_to: BlockNumber,
2513 ) -> ProviderResult<()> {
2514 let shards = self.provider.storage_history_shards(address, storage_key)?;
2515 if shards.is_empty() {
2516 return Ok(());
2517 }
2518
2519 let boundary_idx = shards.iter().position(|(key, _)| {
2522 key.sharded_key.highest_block_number == u64::MAX ||
2523 key.sharded_key.highest_block_number > keep_to
2524 });
2525
2526 let Some(boundary_idx) = boundary_idx else {
2528 let (last_key, last_value) = shards.last().expect("shards is non-empty");
2529 if last_key.sharded_key.highest_block_number != u64::MAX {
2530 self.delete::<tables::StoragesHistory>(last_key.clone())?;
2531 self.put::<tables::StoragesHistory>(
2532 StorageShardedKey::last(address, storage_key),
2533 last_value,
2534 )?;
2535 }
2536 return Ok(());
2537 };
2538
2539 for (key, _) in shards.iter().skip(boundary_idx + 1) {
2541 self.delete::<tables::StoragesHistory>(key.clone())?;
2542 }
2543
2544 let (boundary_key, boundary_list) = &shards[boundary_idx];
2546
2547 self.delete::<tables::StoragesHistory>(boundary_key.clone())?;
2549
2550 let new_last =
2552 BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2553
2554 if new_last.is_empty() {
2555 if boundary_idx == 0 {
2558 return Ok(());
2560 }
2561
2562 let (prev_key, prev_value) = &shards[boundary_idx - 1];
2563 if prev_key.sharded_key.highest_block_number != u64::MAX {
2564 self.delete::<tables::StoragesHistory>(prev_key.clone())?;
2565 self.put::<tables::StoragesHistory>(
2566 StorageShardedKey::last(address, storage_key),
2567 prev_value,
2568 )?;
2569 }
2570 return Ok(());
2571 }
2572
2573 self.put::<tables::StoragesHistory>(
2574 StorageShardedKey::last(address, storage_key),
2575 &new_last,
2576 )?;
2577
2578 Ok(())
2579 }
2580
2581 pub fn clear_account_history(&mut self, address: Address) -> ProviderResult<()> {
2585 let shards = self.provider.account_history_shards(address)?;
2586 for (key, _) in shards {
2587 self.delete::<tables::AccountsHistory>(key)?;
2588 }
2589 Ok(())
2590 }
2591
2592 pub fn clear_storage_history(
2596 &mut self,
2597 address: Address,
2598 storage_key: B256,
2599 ) -> ProviderResult<()> {
2600 let shards = self.provider.storage_history_shards(address, storage_key)?;
2601 for (key, _) in shards {
2602 self.delete::<tables::StoragesHistory>(key)?;
2603 }
2604 Ok(())
2605 }
2606}
2607
2608pub struct RocksTx<'db> {
2618 inner: Transaction<'db, OptimisticTransactionDB>,
2619 provider: &'db RocksDBProvider,
2620}
2621
2622impl fmt::Debug for RocksTx<'_> {
2623 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2624 f.debug_struct("RocksTx").field("provider", &self.provider).finish_non_exhaustive()
2625 }
2626}
2627
2628impl<'db> RocksTx<'db> {
2629 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
2631 let encoded_key = key.encode();
2632 self.get_encoded::<T>(&encoded_key)
2633 }
2634
2635 pub fn get_encoded<T: Table>(
2637 &self,
2638 key: &<T::Key as Encode>::Encoded,
2639 ) -> ProviderResult<Option<T::Value>> {
2640 let cf = self.provider.get_cf_handle::<T>()?;
2641 let result = self.inner.get_cf(cf, key.as_ref()).map_err(|e| {
2642 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2643 message: e.to_string().into(),
2644 code: -1,
2645 }))
2646 })?;
2647
2648 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
2649 }
2650
2651 pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
2653 let encoded_key = key.encode();
2654 self.put_encoded::<T>(&encoded_key, value)
2655 }
2656
2657 pub fn put_encoded<T: Table>(
2659 &self,
2660 key: &<T::Key as Encode>::Encoded,
2661 value: &T::Value,
2662 ) -> ProviderResult<()> {
2663 let cf = self.provider.get_cf_handle::<T>()?;
2664 let mut buf = Vec::new();
2665 let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
2666
2667 self.inner.put_cf(cf, key.as_ref(), value_bytes).map_err(|e| {
2668 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
2669 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
2670 operation: DatabaseWriteOperation::PutUpsert,
2671 table_name: T::NAME,
2672 key: key.as_ref().to_vec(),
2673 })))
2674 })
2675 }
2676
2677 pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
2679 let cf = self.provider.get_cf_handle::<T>()?;
2680 self.inner.delete_cf(cf, key.encode().as_ref()).map_err(|e| {
2681 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
2682 message: e.to_string().into(),
2683 code: -1,
2684 }))
2685 })
2686 }
2687
2688 pub fn iter<T: Table>(&self) -> ProviderResult<RocksTxIter<'_, T>> {
2692 let cf = self.provider.get_cf_handle::<T>()?;
2693 let iter = self.inner.iterator_cf(cf, IteratorMode::Start);
2694 Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2695 }
2696
2697 pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksTxIter<'_, T>> {
2699 let cf = self.provider.get_cf_handle::<T>()?;
2700 let encoded_key = key.encode();
2701 let iter = self
2702 .inner
2703 .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
2704 Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2705 }
2706
2707 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2709 pub fn commit(self) -> ProviderResult<()> {
2710 self.inner.commit().map_err(|e| {
2711 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2712 message: e.to_string().into(),
2713 code: -1,
2714 }))
2715 })
2716 }
2717
2718 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2720 pub fn rollback(self) -> ProviderResult<()> {
2721 self.inner.rollback().map_err(|e| {
2722 ProviderError::Database(DatabaseError::Other(format!("rollback failed: {e}")))
2723 })
2724 }
2725}
2726
2727enum RocksDBIterEnum<'db> {
2729 ReadWrite(rocksdb::DBIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2731 ReadOnly(rocksdb::DBIteratorWithThreadMode<'db, DB>),
2733}
2734
2735impl Iterator for RocksDBIterEnum<'_> {
2736 type Item = Result<(Box<[u8]>, Box<[u8]>), rocksdb::Error>;
2737
2738 fn next(&mut self) -> Option<Self::Item> {
2739 match self {
2740 Self::ReadWrite(iter) => iter.next(),
2741 Self::ReadOnly(iter) => iter.next(),
2742 }
2743 }
2744}
2745
2746enum RocksDBRawIterEnum<'db> {
2751 ReadWrite(DBRawIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2753 ReadOnly(DBRawIteratorWithThreadMode<'db, DB>),
2755}
2756
2757impl RocksDBRawIterEnum<'_> {
2758 fn seek(&mut self, key: impl AsRef<[u8]>) {
2760 match self {
2761 Self::ReadWrite(iter) => iter.seek(key),
2762 Self::ReadOnly(iter) => iter.seek(key),
2763 }
2764 }
2765
2766 fn valid(&self) -> bool {
2768 match self {
2769 Self::ReadWrite(iter) => iter.valid(),
2770 Self::ReadOnly(iter) => iter.valid(),
2771 }
2772 }
2773
2774 fn key(&self) -> Option<&[u8]> {
2776 match self {
2777 Self::ReadWrite(iter) => iter.key(),
2778 Self::ReadOnly(iter) => iter.key(),
2779 }
2780 }
2781
2782 fn value(&self) -> Option<&[u8]> {
2784 match self {
2785 Self::ReadWrite(iter) => iter.value(),
2786 Self::ReadOnly(iter) => iter.value(),
2787 }
2788 }
2789
2790 fn next(&mut self) {
2792 match self {
2793 Self::ReadWrite(iter) => iter.next(),
2794 Self::ReadOnly(iter) => iter.next(),
2795 }
2796 }
2797
2798 fn prev(&mut self) {
2800 match self {
2801 Self::ReadWrite(iter) => iter.prev(),
2802 Self::ReadOnly(iter) => iter.prev(),
2803 }
2804 }
2805
2806 fn status(&self) -> Result<(), rocksdb::Error> {
2808 match self {
2809 Self::ReadWrite(iter) => iter.status(),
2810 Self::ReadOnly(iter) => iter.status(),
2811 }
2812 }
2813}
2814
2815pub struct RocksDBIter<'db, T: Table> {
2819 inner: RocksDBIterEnum<'db>,
2820 _marker: std::marker::PhantomData<T>,
2821}
2822
2823impl<T: Table> fmt::Debug for RocksDBIter<'_, T> {
2824 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2825 f.debug_struct("RocksDBIter").field("table", &T::NAME).finish_non_exhaustive()
2826 }
2827}
2828
2829impl<T: Table> Iterator for RocksDBIter<'_, T> {
2830 type Item = ProviderResult<(T::Key, T::Value)>;
2831
2832 fn next(&mut self) -> Option<Self::Item> {
2833 Some(decode_iter_item::<T>(self.inner.next()?))
2834 }
2835}
2836
2837pub struct RocksDBRawIter<'db> {
2841 inner: RocksDBIterEnum<'db>,
2842}
2843
2844impl fmt::Debug for RocksDBRawIter<'_> {
2845 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2846 f.debug_struct("RocksDBRawIter").finish_non_exhaustive()
2847 }
2848}
2849
2850impl Iterator for RocksDBRawIter<'_> {
2851 type Item = ProviderResult<(Box<[u8]>, Box<[u8]>)>;
2852
2853 fn next(&mut self) -> Option<Self::Item> {
2854 match self.inner.next()? {
2855 Ok(kv) => Some(Ok(kv)),
2856 Err(e) => Some(Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2857 message: e.to_string().into(),
2858 code: -1,
2859 })))),
2860 }
2861 }
2862}
2863
2864pub(crate) struct RocksDBRawKeyIter<'db> {
2866 inner: RocksDBRawIterEnum<'db>,
2867}
2868
2869impl fmt::Debug for RocksDBRawKeyIter<'_> {
2870 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2871 f.debug_struct("RocksDBRawKeyIter").finish_non_exhaustive()
2872 }
2873}
2874
2875impl Iterator for RocksDBRawKeyIter<'_> {
2876 type Item = ProviderResult<Box<[u8]>>;
2877
2878 fn next(&mut self) -> Option<Self::Item> {
2879 if !self.inner.valid() {
2880 return self.inner.status().err().map(|e| {
2881 Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2882 message: e.to_string().into(),
2883 code: -1,
2884 })))
2885 })
2886 }
2887
2888 let Some(key) = self.inner.key() else {
2889 return Some(Err(ProviderError::Database(DatabaseError::Decode)))
2890 };
2891 let key = Box::from(key);
2892 self.inner.next();
2893 Some(Ok(key))
2894 }
2895}
2896
2897pub struct RocksTxIter<'tx, T: Table> {
2901 inner: rocksdb::DBIteratorWithThreadMode<'tx, Transaction<'tx, OptimisticTransactionDB>>,
2902 _marker: std::marker::PhantomData<T>,
2903}
2904
2905impl<T: Table> fmt::Debug for RocksTxIter<'_, T> {
2906 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2907 f.debug_struct("RocksTxIter").field("table", &T::NAME).finish_non_exhaustive()
2908 }
2909}
2910
2911impl<T: Table> Iterator for RocksTxIter<'_, T> {
2912 type Item = ProviderResult<(T::Key, T::Value)>;
2913
2914 fn next(&mut self) -> Option<Self::Item> {
2915 Some(decode_iter_item::<T>(self.inner.next()?))
2916 }
2917}
2918
2919fn decode_iter_item<T: Table>(result: RawKVResult) -> ProviderResult<(T::Key, T::Value)> {
2924 let (key_bytes, value_bytes) = result.map_err(|e| {
2925 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2926 message: e.to_string().into(),
2927 code: -1,
2928 }))
2929 })?;
2930
2931 let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
2932 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2933
2934 let value = T::Value::decompress(&value_bytes)
2935 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2936
2937 Ok((key, value))
2938}
2939
2940const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
2942 match level {
2943 LogLevel::Fatal => rocksdb::LogLevel::Fatal,
2944 LogLevel::Error => rocksdb::LogLevel::Error,
2945 LogLevel::Warn => rocksdb::LogLevel::Warn,
2946 LogLevel::Notice | LogLevel::Verbose => rocksdb::LogLevel::Info,
2947 LogLevel::Debug | LogLevel::Trace | LogLevel::Extra => rocksdb::LogLevel::Debug,
2948 }
2949}
2950
2951fn select_max_open_files() -> i32 {
2968 let file_descriptor_limit = current_file_descriptor_limit();
2969 let max_open_files = max_open_files_for_limit(file_descriptor_limit);
2970
2971 if max_open_files == LIMITED_MAX_OPEN_FILES {
2972 tracing::warn!(
2973 target: "providers::rocksdb",
2974 ?file_descriptor_limit,
2975 threshold = HIGH_FILE_DESCRIPTOR_LIMIT,
2976 max_open_files,
2977 "RocksDB will not keep all files open; performance may be reduced"
2978 );
2979 }
2980
2981 max_open_files
2982}
2983
2984const fn max_open_files_for_limit(file_descriptor_limit: Option<u64>) -> i32 {
2985 match file_descriptor_limit {
2986 Some(limit) if limit >= HIGH_FILE_DESCRIPTOR_LIMIT => KEEP_ALL_FILES_OPEN,
2987 _ => LIMITED_MAX_OPEN_FILES,
2988 }
2989}
2990
2991#[cfg(unix)]
2992#[allow(clippy::useless_conversion)]
2993fn current_file_descriptor_limit() -> Option<u64> {
2994 let mut limit = libc::rlimit { rlim_cur: 0, rlim_max: 0 };
2995 let result = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) };
2997 (result == 0).then(|| limit.rlim_cur.into())
2998}
2999
3000#[cfg(not(unix))]
3001const fn current_file_descriptor_limit() -> Option<u64> {
3002 None
3003}
3004
3005#[cfg(test)]
3006mod tests {
3007 use super::*;
3008 use crate::providers::HistoryInfo;
3009 use alloy_primitives::{Address, Bytes, TxHash, B256};
3010 use reth_db_api::{
3011 models::{
3012 sharded_key::{ShardedKey, NUM_OF_INDICES_IN_SHARD},
3013 storage_sharded_key::StorageShardedKey,
3014 IntegerList,
3015 },
3016 table::Table,
3017 tables,
3018 };
3019 use tempfile::TempDir;
3020
3021 #[test]
3022 fn max_open_files_adapts_to_file_descriptor_limit() {
3023 assert_eq!(max_open_files_for_limit(Some(HIGH_FILE_DESCRIPTOR_LIMIT)), KEEP_ALL_FILES_OPEN);
3024 assert_eq!(
3025 max_open_files_for_limit(Some(HIGH_FILE_DESCRIPTOR_LIMIT - 1)),
3026 LIMITED_MAX_OPEN_FILES
3027 );
3028 assert_eq!(max_open_files_for_limit(None), LIMITED_MAX_OPEN_FILES);
3029 }
3030
3031 #[test]
3032 fn test_with_default_tables_registers_required_column_families() {
3033 let temp_dir = TempDir::new().unwrap();
3034
3035 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3037
3038 let tx_hash = TxHash::from(B256::from([1u8; 32]));
3040 provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3041 assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3042
3043 let key = ShardedKey::new(Address::ZERO, 100);
3045 let value = IntegerList::default();
3046 provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
3047 assert!(provider.get::<tables::AccountsHistory>(key).unwrap().is_some());
3048
3049 let key = StorageShardedKey::new(Address::ZERO, B256::ZERO, 100);
3051 provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
3052 assert!(provider.get::<tables::StoragesHistory>(key).unwrap().is_some());
3053
3054 drop(provider);
3055
3056 let column_families = DB::list_cf(&Options::default(), temp_dir.path()).unwrap();
3057 assert!(!column_families.iter().any(|name| name == tables::BlockAccessLists::NAME));
3058 assert!(!column_families
3059 .iter()
3060 .any(|name| name == tables::BlockAccessListBlockNumbers::NAME));
3061 }
3062
3063 #[test]
3064 fn block_access_lists_store_large_payloads_in_blob_files() {
3065 let temp_dir = TempDir::new().unwrap();
3066 let provider = RocksDBBuilder::new(temp_dir.path())
3067 .with_table::<tables::BlockAccessLists>()
3068 .build()
3069 .unwrap();
3070 let bal_key =
3071 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3072 let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
3073 0;
3074 DEFAULT_BAL_MIN_BLOB_SIZE as usize +
3075 1
3076 ]));
3077
3078 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3079 provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3080
3081 let has_blob_file = std::fs::read_dir(temp_dir.path()).unwrap().any(|entry| {
3082 entry.unwrap().path().extension().is_some_and(|extension| extension == "blob")
3083 });
3084 assert!(has_blob_file);
3085 }
3086
3087 #[derive(Debug)]
3088 struct TestTable;
3089
3090 impl Table for TestTable {
3091 const NAME: &'static str = "TestTable";
3092 const DUPSORT: bool = false;
3093 type Key = u64;
3094 type Value = Vec<u8>;
3095 }
3096
3097 #[test]
3098 fn test_reopens_with_unknown_column_family() {
3099 let temp_dir = TempDir::new().unwrap();
3100 let value = b"test_value".to_vec();
3101
3102 let provider = RocksDBBuilder::new(temp_dir.path())
3103 .with_default_tables()
3104 .with_table::<TestTable>()
3105 .build()
3106 .unwrap();
3107 provider.put::<TestTable>(42, &value).unwrap();
3108 drop(provider);
3109
3110 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3111 assert_eq!(provider.get::<TestTable>(42).unwrap(), Some(value));
3112 }
3113
3114 #[test]
3115 fn test_reopens_blob_column_family_with_legacy_table_set() {
3116 let temp_dir = TempDir::new().unwrap();
3117 let bal_key =
3118 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3119 let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
3120 0;
3121 DEFAULT_BAL_MIN_BLOB_SIZE as usize +
3122 1
3123 ]));
3124
3125 let provider = RocksDBBuilder::new(temp_dir.path())
3126 .with_default_tables()
3127 .with_table::<tables::BlockAccessLists>()
3128 .with_table::<tables::BlockAccessListBlockNumbers>()
3129 .build()
3130 .unwrap();
3131 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3132 provider
3133 .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
3134 .unwrap();
3135 provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3136 drop(provider);
3137
3138 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3139 assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
3140 assert_eq!(
3141 provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
3142 Some(bal_key.number())
3143 );
3144 }
3145
3146 #[test]
3147 fn test_basic_operations() {
3148 let temp_dir = TempDir::new().unwrap();
3149
3150 let provider = RocksDBBuilder::new(temp_dir.path())
3151 .with_table::<TestTable>() .build()
3153 .unwrap();
3154
3155 let key = 42u64;
3156 let value = b"test_value".to_vec();
3157
3158 provider.put::<TestTable>(key, &value).unwrap();
3160
3161 let result = provider.get::<TestTable>(key).unwrap();
3163 assert_eq!(result, Some(value));
3164
3165 provider.delete::<TestTable>(key).unwrap();
3167
3168 assert_eq!(provider.get::<TestTable>(key).unwrap(), None);
3170 }
3171
3172 #[test]
3173 fn test_batch_operations() {
3174 let temp_dir = TempDir::new().unwrap();
3175 let provider =
3176 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3177
3178 provider
3180 .write_batch(|batch| {
3181 for i in 0..10u64 {
3182 let value = format!("value_{i}").into_bytes();
3183 batch.put::<TestTable>(i, &value)?;
3184 }
3185 Ok(())
3186 })
3187 .unwrap();
3188
3189 for i in 0..10u64 {
3191 let value = format!("value_{i}").into_bytes();
3192 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3193 }
3194
3195 provider
3197 .write_batch(|batch| {
3198 for i in 0..10u64 {
3199 batch.delete::<TestTable>(i)?;
3200 }
3201 Ok(())
3202 })
3203 .unwrap();
3204
3205 for i in 0..10u64 {
3207 assert_eq!(provider.get::<TestTable>(i).unwrap(), None);
3208 }
3209 }
3210
3211 #[test]
3212 fn test_with_real_table() {
3213 let temp_dir = TempDir::new().unwrap();
3214 let provider = RocksDBBuilder::new(temp_dir.path())
3215 .with_table::<tables::TransactionHashNumbers>()
3216 .with_metrics()
3217 .build()
3218 .unwrap();
3219
3220 let tx_hash = TxHash::from(B256::from([1u8; 32]));
3221
3222 provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3224 assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3225
3226 provider
3228 .write_batch(|batch| {
3229 for i in 0..10u64 {
3230 let hash = TxHash::from(B256::from([i as u8; 32]));
3231 let value = i * 100;
3232 batch.put::<tables::TransactionHashNumbers>(hash, &value)?;
3233 }
3234 Ok(())
3235 })
3236 .unwrap();
3237
3238 for i in 0..10u64 {
3240 let hash = TxHash::from(B256::from([i as u8; 32]));
3241 assert_eq!(
3242 provider.get::<tables::TransactionHashNumbers>(hash).unwrap(),
3243 Some(i * 100)
3244 );
3245 }
3246 }
3247 #[test]
3248 fn test_statistics_enabled() {
3249 let temp_dir = TempDir::new().unwrap();
3250 let provider = RocksDBBuilder::new(temp_dir.path())
3252 .with_table::<TestTable>()
3253 .with_statistics()
3254 .build()
3255 .unwrap();
3256
3257 for i in 0..10 {
3259 let value = vec![i as u8];
3260 provider.put::<TestTable>(i, &value).unwrap();
3261 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3263 }
3264 }
3265
3266 #[test]
3267 fn test_data_persistence() {
3268 let temp_dir = TempDir::new().unwrap();
3269 let provider =
3270 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3271
3272 let value = vec![42u8; 1000];
3274 for i in 0..100 {
3275 provider.put::<TestTable>(i, &value).unwrap();
3276 }
3277
3278 for i in 0..100 {
3280 assert!(provider.get::<TestTable>(i).unwrap().is_some(), "Data should be readable");
3281 }
3282 }
3283
3284 #[test]
3285 fn test_transaction_read_your_writes() {
3286 let temp_dir = TempDir::new().unwrap();
3287 let provider =
3288 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3289
3290 let tx = provider.tx();
3292
3293 let key = 42u64;
3295 let value = b"test_value".to_vec();
3296 tx.put::<TestTable>(key, &value).unwrap();
3297
3298 let result = tx.get::<TestTable>(key).unwrap();
3300 assert_eq!(
3301 result,
3302 Some(value.clone()),
3303 "Transaction should see its own uncommitted writes"
3304 );
3305
3306 let provider_result = provider.get::<TestTable>(key).unwrap();
3308 assert_eq!(provider_result, None, "Uncommitted data should not be visible outside tx");
3309
3310 tx.commit().unwrap();
3312
3313 let committed_result = provider.get::<TestTable>(key).unwrap();
3315 assert_eq!(committed_result, Some(value), "Committed data should be visible");
3316 }
3317
3318 #[test]
3319 fn test_transaction_rollback() {
3320 let temp_dir = TempDir::new().unwrap();
3321 let provider =
3322 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3323
3324 let key = 100u64;
3326 let initial_value = b"initial".to_vec();
3327 provider.put::<TestTable>(key, &initial_value).unwrap();
3328
3329 let tx = provider.tx();
3331 let new_value = b"modified".to_vec();
3332 tx.put::<TestTable>(key, &new_value).unwrap();
3333
3334 assert_eq!(tx.get::<TestTable>(key).unwrap(), Some(new_value));
3336
3337 tx.rollback().unwrap();
3339
3340 let result = provider.get::<TestTable>(key).unwrap();
3342 assert_eq!(result, Some(initial_value), "Rollback should preserve original data");
3343 }
3344
3345 #[test]
3346 fn test_transaction_iterator() {
3347 let temp_dir = TempDir::new().unwrap();
3348 let provider =
3349 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3350
3351 let tx = provider.tx();
3353
3354 for i in 0..5u64 {
3356 let value = format!("value_{i}").into_bytes();
3357 tx.put::<TestTable>(i, &value).unwrap();
3358 }
3359
3360 let mut count = 0;
3362 for result in tx.iter::<TestTable>().unwrap() {
3363 let (key, value) = result.unwrap();
3364 assert_eq!(value, format!("value_{key}").into_bytes());
3365 count += 1;
3366 }
3367 assert_eq!(count, 5, "Iterator should see all uncommitted writes");
3368
3369 tx.commit().unwrap();
3371 }
3372
3373 #[test]
3374 fn test_batch_manual_commit() {
3375 let temp_dir = TempDir::new().unwrap();
3376 let provider =
3377 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3378
3379 let mut batch = provider.batch();
3381
3382 for i in 0..10u64 {
3384 let value = format!("batch_value_{i}").into_bytes();
3385 batch.put::<TestTable>(i, &value).unwrap();
3386 }
3387
3388 assert_eq!(batch.len(), 10);
3390 assert!(!batch.is_empty());
3391
3392 assert_eq!(provider.get::<TestTable>(0).unwrap(), None);
3394
3395 batch.commit().unwrap();
3397
3398 for i in 0..10u64 {
3400 let value = format!("batch_value_{i}").into_bytes();
3401 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3402 }
3403 }
3404
3405 #[test]
3406 fn test_first_and_last_entry() {
3407 let temp_dir = TempDir::new().unwrap();
3408 let provider =
3409 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3410
3411 assert_eq!(provider.first::<TestTable>().unwrap(), None);
3413 assert_eq!(provider.last::<TestTable>().unwrap(), None);
3414
3415 provider.put::<TestTable>(10, &b"value_10".to_vec()).unwrap();
3417 provider.put::<TestTable>(20, &b"value_20".to_vec()).unwrap();
3418 provider.put::<TestTable>(5, &b"value_5".to_vec()).unwrap();
3419
3420 let first = provider.first::<TestTable>().unwrap();
3422 assert_eq!(first, Some((5, b"value_5".to_vec())));
3423
3424 let last = provider.last::<TestTable>().unwrap();
3426 assert_eq!(last, Some((20, b"value_20".to_vec())));
3427 }
3428
3429 #[test]
3434 fn test_account_history_info_pruned_before_first_entry() {
3435 let temp_dir = TempDir::new().unwrap();
3436 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3437
3438 let address = Address::from([0x42; 20]);
3439
3440 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3442 let shard_key = ShardedKey::new(address, u64::MAX);
3443 provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3444
3445 let result =
3450 provider.snapshot().account_history_info(address, 50, Some(100), u64::MAX).unwrap();
3451 assert_eq!(result, HistoryInfo::InChangeset(100));
3452 }
3453
3454 #[test]
3456 fn test_account_history_info_read_only_and_catch_up() {
3457 let temp_dir = TempDir::new().unwrap();
3458 let address = Address::from([0x42; 20]);
3459 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3460 let shard_key = ShardedKey::new(address, u64::MAX);
3461
3462 let rw_provider =
3464 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3465 rw_provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3466
3467 let ro_provider = RocksDBBuilder::new(temp_dir.path())
3469 .with_default_tables()
3470 .with_read_only(true)
3471 .build()
3472 .unwrap();
3473
3474 let result =
3475 ro_provider.snapshot().account_history_info(address, 200, None, u64::MAX).unwrap();
3476 assert_eq!(result, HistoryInfo::InChangeset(200));
3477
3478 let result =
3479 ro_provider.snapshot().account_history_info(address, 50, None, u64::MAX).unwrap();
3480 assert_eq!(result, HistoryInfo::NotYetWritten);
3481
3482 let result =
3483 ro_provider.snapshot().account_history_info(address, 400, None, u64::MAX).unwrap();
3484 assert_eq!(result, HistoryInfo::InPlainState);
3485
3486 let address2 = Address::from([0x43; 20]);
3488 let chunk2 = IntegerList::new([500, 600]).unwrap();
3489 let shard_key2 = ShardedKey::new(address2, u64::MAX);
3490 rw_provider.put::<tables::AccountsHistory>(shard_key2, &chunk2).unwrap();
3491
3492 let result =
3494 ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3495 assert_eq!(result, HistoryInfo::NotYetWritten);
3496
3497 ro_provider.try_catch_up_with_primary().unwrap();
3499
3500 let result =
3501 ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3502 assert_eq!(result, HistoryInfo::InChangeset(500));
3503 }
3504
3505 #[test]
3506 fn test_account_history_info_ignores_blocks_above_visible_tip() {
3507 let temp_dir = TempDir::new().unwrap();
3508 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3509
3510 let address = Address::from([0x42; 20]);
3511
3512 provider
3513 .put::<tables::AccountsHistory>(
3514 ShardedKey::new(address, 110),
3515 &IntegerList::new([100, 110]).unwrap(),
3516 )
3517 .unwrap();
3518 provider
3519 .put::<tables::AccountsHistory>(
3520 ShardedKey::new(address, u64::MAX),
3521 &IntegerList::new([200, 210]).unwrap(),
3522 )
3523 .unwrap();
3524
3525 let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3526 assert_eq!(result, HistoryInfo::InPlainState);
3527 }
3528
3529 #[test]
3530 fn test_account_history_info_mixed_shard_respects_visible_tip() {
3531 let temp_dir = TempDir::new().unwrap();
3532 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3533
3534 let address = Address::from([0x42; 20]);
3535 provider
3536 .put::<tables::AccountsHistory>(
3537 ShardedKey::new(address, u64::MAX),
3538 &IntegerList::new([100, 150, 300]).unwrap(),
3539 )
3540 .unwrap();
3541
3542 let result = provider.snapshot().account_history_info(address, 120, None, 200).unwrap();
3543 assert_eq!(result, HistoryInfo::InChangeset(150));
3544
3545 let result = provider.snapshot().account_history_info(address, 201, None, 200).unwrap();
3546 assert_eq!(result, HistoryInfo::InPlainState);
3547 }
3548
3549 #[test]
3550 fn test_account_history_info_only_stale_entries_use_fallback() {
3551 let temp_dir = TempDir::new().unwrap();
3552 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3553
3554 let address = Address::from([0x42; 20]);
3555 provider
3556 .put::<tables::AccountsHistory>(
3557 ShardedKey::new(address, u64::MAX),
3558 &IntegerList::new([200, 210]).unwrap(),
3559 )
3560 .unwrap();
3561
3562 let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3563 assert_eq!(result, HistoryInfo::NotYetWritten);
3564
3565 let result =
3566 provider.snapshot().account_history_info(address, 150, Some(100), 150).unwrap();
3567 assert_eq!(result, HistoryInfo::MaybeInPlainState);
3568 }
3569
3570 #[test]
3571 fn test_account_history_shard_split_at_boundary() {
3572 let temp_dir = TempDir::new().unwrap();
3573 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3574
3575 let address = Address::from([0x42; 20]);
3576 let limit = NUM_OF_INDICES_IN_SHARD;
3577
3578 let indices: Vec<u64> = (0..=(limit as u64)).collect();
3580 let mut batch = provider.batch();
3581 batch.append_account_history_shard(address, indices).unwrap();
3582 batch.commit().unwrap();
3583
3584 let completed_key = ShardedKey::new(address, (limit - 1) as u64);
3586 let sentinel_key = ShardedKey::new(address, u64::MAX);
3587
3588 let completed_shard = provider.get::<tables::AccountsHistory>(completed_key).unwrap();
3589 let sentinel_shard = provider.get::<tables::AccountsHistory>(sentinel_key).unwrap();
3590
3591 assert!(completed_shard.is_some(), "completed shard should exist");
3592 assert!(sentinel_shard.is_some(), "sentinel shard should exist");
3593
3594 let completed_shard = completed_shard.unwrap();
3595 let sentinel_shard = sentinel_shard.unwrap();
3596
3597 assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
3598 assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
3599 }
3600
3601 #[test]
3602 fn test_account_history_multiple_shard_splits() {
3603 let temp_dir = TempDir::new().unwrap();
3604 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3605
3606 let address = Address::from([0x43; 20]);
3607 let limit = NUM_OF_INDICES_IN_SHARD;
3608
3609 let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
3611 let mut batch = provider.batch();
3612 batch.append_account_history_shard(address, first_batch_indices).unwrap();
3613 batch.commit().unwrap();
3614
3615 let sentinel_key = ShardedKey::new(address, u64::MAX);
3617 let shard = provider.get::<tables::AccountsHistory>(sentinel_key.clone()).unwrap();
3618 assert!(shard.is_some());
3619 assert_eq!(shard.unwrap().len(), limit as u64);
3620
3621 let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
3623 let mut batch = provider.batch();
3624 batch.append_account_history_shard(address, second_batch_indices).unwrap();
3625 batch.commit().unwrap();
3626
3627 let first_completed = ShardedKey::new(address, (limit - 1) as u64);
3629 let second_completed = ShardedKey::new(address, (2 * limit - 1) as u64);
3630
3631 assert!(
3632 provider.get::<tables::AccountsHistory>(first_completed).unwrap().is_some(),
3633 "first completed shard should exist"
3634 );
3635 assert!(
3636 provider.get::<tables::AccountsHistory>(second_completed).unwrap().is_some(),
3637 "second completed shard should exist"
3638 );
3639 assert!(
3640 provider.get::<tables::AccountsHistory>(sentinel_key).unwrap().is_some(),
3641 "sentinel shard should exist"
3642 );
3643 }
3644
3645 #[test]
3646 fn test_storage_history_shard_split_at_boundary() {
3647 let temp_dir = TempDir::new().unwrap();
3648 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3649
3650 let address = Address::from([0x44; 20]);
3651 let slot = B256::from([0x55; 32]);
3652 let limit = NUM_OF_INDICES_IN_SHARD;
3653
3654 let indices: Vec<u64> = (0..=(limit as u64)).collect();
3656 let mut batch = provider.batch();
3657 batch.append_storage_history_shard(address, slot, indices).unwrap();
3658 batch.commit().unwrap();
3659
3660 let completed_key = StorageShardedKey::new(address, slot, (limit - 1) as u64);
3662 let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
3663
3664 let completed_shard = provider.get::<tables::StoragesHistory>(completed_key).unwrap();
3665 let sentinel_shard = provider.get::<tables::StoragesHistory>(sentinel_key).unwrap();
3666
3667 assert!(completed_shard.is_some(), "completed shard should exist");
3668 assert!(sentinel_shard.is_some(), "sentinel shard should exist");
3669
3670 let completed_shard = completed_shard.unwrap();
3671 let sentinel_shard = sentinel_shard.unwrap();
3672
3673 assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
3674 assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
3675 }
3676
3677 #[test]
3678 fn test_storage_history_multiple_shard_splits() {
3679 let temp_dir = TempDir::new().unwrap();
3680 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3681
3682 let address = Address::from([0x46; 20]);
3683 let slot = B256::from([0x57; 32]);
3684 let limit = NUM_OF_INDICES_IN_SHARD;
3685
3686 let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
3688 let mut batch = provider.batch();
3689 batch.append_storage_history_shard(address, slot, first_batch_indices).unwrap();
3690 batch.commit().unwrap();
3691
3692 let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
3694 let shard = provider.get::<tables::StoragesHistory>(sentinel_key.clone()).unwrap();
3695 assert!(shard.is_some());
3696 assert_eq!(shard.unwrap().len(), limit as u64);
3697
3698 let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
3700 let mut batch = provider.batch();
3701 batch.append_storage_history_shard(address, slot, second_batch_indices).unwrap();
3702 batch.commit().unwrap();
3703
3704 let first_completed = StorageShardedKey::new(address, slot, (limit - 1) as u64);
3706 let second_completed = StorageShardedKey::new(address, slot, (2 * limit - 1) as u64);
3707
3708 assert!(
3709 provider.get::<tables::StoragesHistory>(first_completed).unwrap().is_some(),
3710 "first completed shard should exist"
3711 );
3712 assert!(
3713 provider.get::<tables::StoragesHistory>(second_completed).unwrap().is_some(),
3714 "second completed shard should exist"
3715 );
3716 assert!(
3717 provider.get::<tables::StoragesHistory>(sentinel_key).unwrap().is_some(),
3718 "sentinel shard should exist"
3719 );
3720 }
3721
3722 #[test]
3723 fn test_clear_table() {
3724 let temp_dir = TempDir::new().unwrap();
3725 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3726
3727 let address = Address::from([0x42; 20]);
3728 let key = ShardedKey::new(address, u64::MAX);
3729 let blocks = BlockNumberList::new_pre_sorted([1, 2, 3]);
3730
3731 provider.put::<tables::AccountsHistory>(key.clone(), &blocks).unwrap();
3732 assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
3733
3734 provider.clear::<tables::AccountsHistory>().unwrap();
3735
3736 assert!(
3737 provider.get::<tables::AccountsHistory>(key).unwrap().is_none(),
3738 "table should be empty after clear"
3739 );
3740 assert!(
3741 provider.first::<tables::AccountsHistory>().unwrap().is_none(),
3742 "first() should return None after clear"
3743 );
3744 }
3745
3746 #[test]
3747 fn test_clear_empty_table() {
3748 let temp_dir = TempDir::new().unwrap();
3749 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3750
3751 assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
3752
3753 provider.clear::<tables::AccountsHistory>().unwrap();
3754
3755 assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
3756 }
3757
3758 #[test]
3759 fn test_unwind_account_history_to_basic() {
3760 let temp_dir = TempDir::new().unwrap();
3761 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3762
3763 let address = Address::from([0x42; 20]);
3764
3765 let mut batch = provider.batch();
3767 batch.append_account_history_shard(address, 0..=10).unwrap();
3768 batch.commit().unwrap();
3769
3770 let key = ShardedKey::new(address, u64::MAX);
3772 let result = provider.get::<tables::AccountsHistory>(key.clone()).unwrap();
3773 assert!(result.is_some());
3774 let blocks: Vec<u64> = result.unwrap().iter().collect();
3775 assert_eq!(blocks, (0..=10).collect::<Vec<_>>());
3776
3777 let mut batch = provider.batch();
3779 batch.unwind_account_history_to(address, 5).unwrap();
3780 batch.commit().unwrap();
3781
3782 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3784 assert!(result.is_some());
3785 let blocks: Vec<u64> = result.unwrap().iter().collect();
3786 assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
3787 }
3788
3789 #[test]
3790 fn test_unwind_account_history_to_removes_all() {
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::from([0x42; 20]);
3795
3796 let mut batch = provider.batch();
3798 batch.append_account_history_shard(address, 5..=10).unwrap();
3799 batch.commit().unwrap();
3800
3801 let mut batch = provider.batch();
3803 batch.unwind_account_history_to(address, 4).unwrap();
3804 batch.commit().unwrap();
3805
3806 let key = ShardedKey::new(address, u64::MAX);
3808 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3809 assert!(result.is_none(), "Should have no data after full unwind");
3810 }
3811
3812 #[test]
3813 fn test_unwind_account_history_to_no_op() {
3814 let temp_dir = TempDir::new().unwrap();
3815 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3816
3817 let address = Address::from([0x42; 20]);
3818
3819 let mut batch = provider.batch();
3821 batch.append_account_history_shard(address, 0..=5).unwrap();
3822 batch.commit().unwrap();
3823
3824 let mut batch = provider.batch();
3826 batch.unwind_account_history_to(address, 10).unwrap();
3827 batch.commit().unwrap();
3828
3829 let key = ShardedKey::new(address, u64::MAX);
3831 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3832 assert!(result.is_some());
3833 let blocks: Vec<u64> = result.unwrap().iter().collect();
3834 assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
3835 }
3836
3837 #[test]
3838 fn test_unwind_account_history_to_block_zero() {
3839 let temp_dir = TempDir::new().unwrap();
3840 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3841
3842 let address = Address::from([0x42; 20]);
3843
3844 let mut batch = provider.batch();
3846 batch.append_account_history_shard(address, 0..=5).unwrap();
3847 batch.commit().unwrap();
3848
3849 let mut batch = provider.batch();
3852 batch.unwind_account_history_to(address, 0).unwrap();
3853 batch.commit().unwrap();
3854
3855 let key = ShardedKey::new(address, u64::MAX);
3857 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3858 assert!(result.is_some());
3859 let blocks: Vec<u64> = result.unwrap().iter().collect();
3860 assert_eq!(blocks, vec![0]);
3861 }
3862
3863 #[test]
3864 fn test_unwind_account_history_to_multi_shard() {
3865 let temp_dir = TempDir::new().unwrap();
3866 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3867
3868 let address = Address::from([0x42; 20]);
3869
3870 let mut batch = provider.batch();
3873
3874 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
3876 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
3877
3878 let shard2 = BlockNumberList::new_pre_sorted(51..=100);
3880 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
3881
3882 batch.commit().unwrap();
3883
3884 let shards = provider.account_history_shards(address).unwrap();
3886 assert_eq!(shards.len(), 2);
3887
3888 let mut batch = provider.batch();
3890 batch.unwind_account_history_to(address, 75).unwrap();
3891 batch.commit().unwrap();
3892
3893 let shards = provider.account_history_shards(address).unwrap();
3895 assert_eq!(shards.len(), 2);
3896
3897 assert_eq!(shards[0].0.highest_block_number, 50);
3899 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
3900
3901 assert_eq!(shards[1].0.highest_block_number, u64::MAX);
3903 assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
3904 }
3905
3906 #[test]
3907 fn test_unwind_account_history_to_multi_shard_boundary_empty() {
3908 let temp_dir = TempDir::new().unwrap();
3909 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3910
3911 let address = Address::from([0x42; 20]);
3912
3913 let mut batch = provider.batch();
3915
3916 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
3918 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
3919
3920 let shard2 = BlockNumberList::new_pre_sorted(75..=100);
3922 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
3923
3924 batch.commit().unwrap();
3925
3926 let mut batch = provider.batch();
3928 batch.unwind_account_history_to(address, 60).unwrap();
3929 batch.commit().unwrap();
3930
3931 let shards = provider.account_history_shards(address).unwrap();
3933 assert_eq!(shards.len(), 1);
3934 assert_eq!(shards[0].0.highest_block_number, u64::MAX);
3935 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
3936 }
3937
3938 #[test]
3939 fn test_account_history_shards_iterator() {
3940 let temp_dir = TempDir::new().unwrap();
3941 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3942
3943 let address = Address::from([0x42; 20]);
3944 let other_address = Address::from([0x43; 20]);
3945
3946 let mut batch = provider.batch();
3948 batch.append_account_history_shard(address, 0..=5).unwrap();
3949 batch.append_account_history_shard(other_address, 10..=15).unwrap();
3950 batch.commit().unwrap();
3951
3952 let shards = provider.account_history_shards(address).unwrap();
3954 assert_eq!(shards.len(), 1);
3955 assert_eq!(shards[0].0.key, address);
3956
3957 let shards = provider.account_history_shards(other_address).unwrap();
3959 assert_eq!(shards.len(), 1);
3960 assert_eq!(shards[0].0.key, other_address);
3961
3962 let non_existent = Address::from([0x99; 20]);
3964 let shards = provider.account_history_shards(non_existent).unwrap();
3965 assert!(shards.is_empty());
3966 }
3967
3968 #[test]
3969 fn test_clear_account_history() {
3970 let temp_dir = TempDir::new().unwrap();
3971 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3972
3973 let address = Address::from([0x42; 20]);
3974
3975 let mut batch = provider.batch();
3977 batch.append_account_history_shard(address, 0..=10).unwrap();
3978 batch.commit().unwrap();
3979
3980 let mut batch = provider.batch();
3982 batch.clear_account_history(address).unwrap();
3983 batch.commit().unwrap();
3984
3985 let shards = provider.account_history_shards(address).unwrap();
3987 assert!(shards.is_empty(), "All shards should be deleted");
3988 }
3989
3990 #[test]
3991 fn test_unwind_non_sentinel_boundary() {
3992 let temp_dir = TempDir::new().unwrap();
3993 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3994
3995 let address = Address::from([0x42; 20]);
3996
3997 let mut batch = provider.batch();
3999
4000 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4002 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4003
4004 let shard2 = BlockNumberList::new_pre_sorted(51..=100);
4006 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 100), &shard2).unwrap();
4007
4008 let shard3 = BlockNumberList::new_pre_sorted(101..=150);
4010 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard3).unwrap();
4011
4012 batch.commit().unwrap();
4013
4014 let shards = provider.account_history_shards(address).unwrap();
4016 assert_eq!(shards.len(), 3);
4017
4018 let mut batch = provider.batch();
4020 batch.unwind_account_history_to(address, 75).unwrap();
4021 batch.commit().unwrap();
4022
4023 let shards = provider.account_history_shards(address).unwrap();
4025 assert_eq!(shards.len(), 2);
4026
4027 assert_eq!(shards[0].0.highest_block_number, 50);
4029 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4030
4031 assert_eq!(shards[1].0.highest_block_number, u64::MAX);
4033 assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
4034 }
4035
4036 #[test]
4037 fn test_batch_auto_commit_on_threshold() {
4038 let temp_dir = TempDir::new().unwrap();
4039 let provider =
4040 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
4041
4042 let mut batch = RocksDBBatch {
4044 provider: &provider,
4045 inner: WriteBatchWithTransaction::<true>::default(),
4046 buf: Vec::new(),
4047 auto_commit_threshold: Some(1024), };
4049
4050 for i in 0..100u64 {
4053 let value = format!("value_{i:04}").into_bytes();
4054 batch.put::<TestTable>(i, &value).unwrap();
4055 }
4056
4057 let first_visible = provider.get::<TestTable>(0).unwrap();
4060 assert!(first_visible.is_some(), "Auto-committed data should be visible");
4061
4062 batch.commit().unwrap();
4064
4065 for i in 0..100u64 {
4067 let value = format!("value_{i:04}").into_bytes();
4068 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
4069 }
4070 }
4071
4072 struct AccountPruneCase {
4076 name: &'static str,
4077 initial_shards: &'static [(u64, &'static [u64])],
4078 prune_to: u64,
4079 expected_outcome: PruneShardOutcome,
4080 expected_shards: &'static [(u64, &'static [u64])],
4081 }
4082
4083 struct StoragePruneCase {
4085 name: &'static str,
4086 initial_shards: &'static [(u64, &'static [u64])],
4087 prune_to: u64,
4088 expected_outcome: PruneShardOutcome,
4089 expected_shards: &'static [(u64, &'static [u64])],
4090 }
4091
4092 #[test]
4093 fn test_prune_account_history_cases() {
4094 const MAX: u64 = u64::MAX;
4095 const CASES: &[AccountPruneCase] = &[
4096 AccountPruneCase {
4097 name: "single_shard_truncate",
4098 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4099 prune_to: 25,
4100 expected_outcome: PruneShardOutcome::Updated,
4101 expected_shards: &[(MAX, &[30, 40])],
4102 },
4103 AccountPruneCase {
4104 name: "single_shard_delete_all",
4105 initial_shards: &[(MAX, &[10, 20])],
4106 prune_to: 20,
4107 expected_outcome: PruneShardOutcome::Deleted,
4108 expected_shards: &[],
4109 },
4110 AccountPruneCase {
4111 name: "single_shard_noop",
4112 initial_shards: &[(MAX, &[10, 20])],
4113 prune_to: 5,
4114 expected_outcome: PruneShardOutcome::Unchanged,
4115 expected_shards: &[(MAX, &[10, 20])],
4116 },
4117 AccountPruneCase {
4118 name: "no_shards",
4119 initial_shards: &[],
4120 prune_to: 100,
4121 expected_outcome: PruneShardOutcome::Unchanged,
4122 expected_shards: &[],
4123 },
4124 AccountPruneCase {
4125 name: "multi_shard_truncate_first",
4126 initial_shards: &[(30, &[10, 20, 30]), (MAX, &[40, 50, 60])],
4127 prune_to: 25,
4128 expected_outcome: PruneShardOutcome::Updated,
4129 expected_shards: &[(30, &[30]), (MAX, &[40, 50, 60])],
4130 },
4131 AccountPruneCase {
4132 name: "delete_first_shard_sentinel_unchanged",
4133 initial_shards: &[(20, &[10, 20]), (MAX, &[30, 40])],
4134 prune_to: 20,
4135 expected_outcome: PruneShardOutcome::Deleted,
4136 expected_shards: &[(MAX, &[30, 40])],
4137 },
4138 AccountPruneCase {
4139 name: "multi_shard_delete_all_but_last",
4140 initial_shards: &[(10, &[5, 10]), (20, &[15, 20]), (MAX, &[25, 30])],
4141 prune_to: 22,
4142 expected_outcome: PruneShardOutcome::Deleted,
4143 expected_shards: &[(MAX, &[25, 30])],
4144 },
4145 AccountPruneCase {
4146 name: "mid_shard_preserves_key",
4147 initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4148 prune_to: 25,
4149 expected_outcome: PruneShardOutcome::Updated,
4150 expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4151 },
4152 AccountPruneCase {
4154 name: "equiv_delete_early_shards_keep_sentinel",
4155 initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4156 prune_to: 55,
4157 expected_outcome: PruneShardOutcome::Deleted,
4158 expected_shards: &[(MAX, &[60, 70])],
4159 },
4160 AccountPruneCase {
4161 name: "equiv_sentinel_becomes_empty_with_prev",
4162 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4163 prune_to: 40,
4164 expected_outcome: PruneShardOutcome::Deleted,
4165 expected_shards: &[(MAX, &[50])],
4166 },
4167 AccountPruneCase {
4168 name: "equiv_all_shards_become_empty",
4169 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4170 prune_to: 51,
4171 expected_outcome: PruneShardOutcome::Deleted,
4172 expected_shards: &[],
4173 },
4174 AccountPruneCase {
4175 name: "equiv_non_sentinel_last_shard_promoted",
4176 initial_shards: &[(100, &[50, 75, 100])],
4177 prune_to: 60,
4178 expected_outcome: PruneShardOutcome::Updated,
4179 expected_shards: &[(MAX, &[75, 100])],
4180 },
4181 AccountPruneCase {
4182 name: "equiv_filter_within_shard",
4183 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4184 prune_to: 25,
4185 expected_outcome: PruneShardOutcome::Updated,
4186 expected_shards: &[(MAX, &[30, 40])],
4187 },
4188 AccountPruneCase {
4189 name: "equiv_multi_shard_partial_delete",
4190 initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4191 prune_to: 35,
4192 expected_outcome: PruneShardOutcome::Deleted,
4193 expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4194 },
4195 ];
4196
4197 let address = Address::from([0x42; 20]);
4198
4199 for case in CASES {
4200 let temp_dir = TempDir::new().unwrap();
4201 let provider =
4202 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4203
4204 let mut batch = provider.batch();
4206 for (highest, blocks) in case.initial_shards {
4207 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4208 batch
4209 .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4210 .unwrap();
4211 }
4212 batch.commit().unwrap();
4213
4214 let mut batch = provider.batch();
4216 let outcome = batch.prune_account_history_to(address, case.prune_to).unwrap();
4217 batch.commit().unwrap();
4218
4219 assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4221
4222 let shards = provider.account_history_shards(address).unwrap();
4224 assert_eq!(
4225 shards.len(),
4226 case.expected_shards.len(),
4227 "case '{}': wrong shard count",
4228 case.name
4229 );
4230 for (i, ((key, blocks), (exp_key, exp_blocks))) in
4231 shards.iter().zip(case.expected_shards.iter()).enumerate()
4232 {
4233 assert_eq!(
4234 key.highest_block_number, *exp_key,
4235 "case '{}': shard {} wrong key",
4236 case.name, i
4237 );
4238 assert_eq!(
4239 blocks.iter().collect::<Vec<_>>(),
4240 *exp_blocks,
4241 "case '{}': shard {} wrong blocks",
4242 case.name,
4243 i
4244 );
4245 }
4246 }
4247 }
4248
4249 #[test]
4250 fn test_prune_storage_history_cases() {
4251 const MAX: u64 = u64::MAX;
4252 const CASES: &[StoragePruneCase] = &[
4253 StoragePruneCase {
4254 name: "single_shard_truncate",
4255 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4256 prune_to: 25,
4257 expected_outcome: PruneShardOutcome::Updated,
4258 expected_shards: &[(MAX, &[30, 40])],
4259 },
4260 StoragePruneCase {
4261 name: "single_shard_delete_all",
4262 initial_shards: &[(MAX, &[10, 20])],
4263 prune_to: 20,
4264 expected_outcome: PruneShardOutcome::Deleted,
4265 expected_shards: &[],
4266 },
4267 StoragePruneCase {
4268 name: "noop",
4269 initial_shards: &[(MAX, &[10, 20])],
4270 prune_to: 5,
4271 expected_outcome: PruneShardOutcome::Unchanged,
4272 expected_shards: &[(MAX, &[10, 20])],
4273 },
4274 StoragePruneCase {
4275 name: "no_shards",
4276 initial_shards: &[],
4277 prune_to: 100,
4278 expected_outcome: PruneShardOutcome::Unchanged,
4279 expected_shards: &[],
4280 },
4281 StoragePruneCase {
4282 name: "mid_shard_preserves_key",
4283 initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4284 prune_to: 25,
4285 expected_outcome: PruneShardOutcome::Updated,
4286 expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4287 },
4288 StoragePruneCase {
4290 name: "equiv_sentinel_promotion",
4291 initial_shards: &[(100, &[50, 75, 100])],
4292 prune_to: 60,
4293 expected_outcome: PruneShardOutcome::Updated,
4294 expected_shards: &[(MAX, &[75, 100])],
4295 },
4296 StoragePruneCase {
4297 name: "equiv_delete_early_shards_keep_sentinel",
4298 initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4299 prune_to: 55,
4300 expected_outcome: PruneShardOutcome::Deleted,
4301 expected_shards: &[(MAX, &[60, 70])],
4302 },
4303 StoragePruneCase {
4304 name: "equiv_sentinel_becomes_empty_with_prev",
4305 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4306 prune_to: 40,
4307 expected_outcome: PruneShardOutcome::Deleted,
4308 expected_shards: &[(MAX, &[50])],
4309 },
4310 StoragePruneCase {
4311 name: "equiv_all_shards_become_empty",
4312 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4313 prune_to: 51,
4314 expected_outcome: PruneShardOutcome::Deleted,
4315 expected_shards: &[],
4316 },
4317 StoragePruneCase {
4318 name: "equiv_filter_within_shard",
4319 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4320 prune_to: 25,
4321 expected_outcome: PruneShardOutcome::Updated,
4322 expected_shards: &[(MAX, &[30, 40])],
4323 },
4324 StoragePruneCase {
4325 name: "equiv_multi_shard_partial_delete",
4326 initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4327 prune_to: 35,
4328 expected_outcome: PruneShardOutcome::Deleted,
4329 expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4330 },
4331 ];
4332
4333 let address = Address::from([0x42; 20]);
4334 let storage_key = B256::from([0x01; 32]);
4335
4336 for case in CASES {
4337 let temp_dir = TempDir::new().unwrap();
4338 let provider =
4339 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4340
4341 let mut batch = provider.batch();
4343 for (highest, blocks) in case.initial_shards {
4344 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4345 let key = if *highest == MAX {
4346 StorageShardedKey::last(address, storage_key)
4347 } else {
4348 StorageShardedKey::new(address, storage_key, *highest)
4349 };
4350 batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4351 }
4352 batch.commit().unwrap();
4353
4354 let mut batch = provider.batch();
4356 let outcome =
4357 batch.prune_storage_history_to(address, storage_key, case.prune_to).unwrap();
4358 batch.commit().unwrap();
4359
4360 assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4362
4363 let shards = provider.storage_history_shards(address, storage_key).unwrap();
4365 assert_eq!(
4366 shards.len(),
4367 case.expected_shards.len(),
4368 "case '{}': wrong shard count",
4369 case.name
4370 );
4371 for (i, ((key, blocks), (exp_key, exp_blocks))) in
4372 shards.iter().zip(case.expected_shards.iter()).enumerate()
4373 {
4374 assert_eq!(
4375 key.sharded_key.highest_block_number, *exp_key,
4376 "case '{}': shard {} wrong key",
4377 case.name, i
4378 );
4379 assert_eq!(
4380 blocks.iter().collect::<Vec<_>>(),
4381 *exp_blocks,
4382 "case '{}': shard {} wrong blocks",
4383 case.name,
4384 i
4385 );
4386 }
4387 }
4388 }
4389
4390 #[test]
4391 fn test_prune_storage_history_does_not_affect_other_slots() {
4392 let temp_dir = TempDir::new().unwrap();
4393 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4394
4395 let address = Address::from([0x42; 20]);
4396 let slot1 = B256::from([0x01; 32]);
4397 let slot2 = B256::from([0x02; 32]);
4398
4399 let mut batch = provider.batch();
4401 batch
4402 .put::<tables::StoragesHistory>(
4403 StorageShardedKey::last(address, slot1),
4404 &BlockNumberList::new_pre_sorted([10u64, 20]),
4405 )
4406 .unwrap();
4407 batch
4408 .put::<tables::StoragesHistory>(
4409 StorageShardedKey::last(address, slot2),
4410 &BlockNumberList::new_pre_sorted([30u64, 40]),
4411 )
4412 .unwrap();
4413 batch.commit().unwrap();
4414
4415 let mut batch = provider.batch();
4417 let outcome = batch.prune_storage_history_to(address, slot1, 20).unwrap();
4418 batch.commit().unwrap();
4419
4420 assert_eq!(outcome, PruneShardOutcome::Deleted);
4421
4422 let shards1 = provider.storage_history_shards(address, slot1).unwrap();
4424 assert!(shards1.is_empty());
4425
4426 let shards2 = provider.storage_history_shards(address, slot2).unwrap();
4428 assert_eq!(shards2.len(), 1);
4429 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![30, 40]);
4430 }
4431
4432 #[test]
4433 fn test_prune_invariants() {
4434 let address = Address::from([0x42; 20]);
4436 let storage_key = B256::from([0x01; 32]);
4437
4438 #[expect(clippy::type_complexity)]
4440 let invariant_cases: &[(&[(u64, &[u64])], u64)] = &[
4441 (&[(10, &[5, 10]), (20, &[15, 20]), (u64::MAX, &[25, 30])], 20),
4443 (&[(100, &[50, 100])], 60),
4445 ];
4446
4447 for (initial_shards, prune_to) in invariant_cases {
4448 {
4450 let temp_dir = TempDir::new().unwrap();
4451 let provider =
4452 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4453
4454 let mut batch = provider.batch();
4455 for (highest, blocks) in *initial_shards {
4456 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4457 batch
4458 .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4459 .unwrap();
4460 }
4461 batch.commit().unwrap();
4462
4463 let mut batch = provider.batch();
4464 batch.prune_account_history_to(address, *prune_to).unwrap();
4465 batch.commit().unwrap();
4466
4467 let shards = provider.account_history_shards(address).unwrap();
4468
4469 for (key, blocks) in &shards {
4471 assert!(
4472 !blocks.is_empty(),
4473 "Account: empty shard at key {}",
4474 key.highest_block_number
4475 );
4476 }
4477
4478 if !shards.is_empty() {
4480 let last = shards.last().unwrap();
4481 assert_eq!(
4482 last.0.highest_block_number,
4483 u64::MAX,
4484 "Account: last shard must be sentinel"
4485 );
4486 }
4487 }
4488
4489 {
4491 let temp_dir = TempDir::new().unwrap();
4492 let provider =
4493 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4494
4495 let mut batch = provider.batch();
4496 for (highest, blocks) in *initial_shards {
4497 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4498 let key = if *highest == u64::MAX {
4499 StorageShardedKey::last(address, storage_key)
4500 } else {
4501 StorageShardedKey::new(address, storage_key, *highest)
4502 };
4503 batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4504 }
4505 batch.commit().unwrap();
4506
4507 let mut batch = provider.batch();
4508 batch.prune_storage_history_to(address, storage_key, *prune_to).unwrap();
4509 batch.commit().unwrap();
4510
4511 let shards = provider.storage_history_shards(address, storage_key).unwrap();
4512
4513 for (key, blocks) in &shards {
4515 assert!(
4516 !blocks.is_empty(),
4517 "Storage: empty shard at key {}",
4518 key.sharded_key.highest_block_number
4519 );
4520 }
4521
4522 if !shards.is_empty() {
4524 let last = shards.last().unwrap();
4525 assert_eq!(
4526 last.0.sharded_key.highest_block_number,
4527 u64::MAX,
4528 "Storage: last shard must be sentinel"
4529 );
4530 }
4531 }
4532 }
4533 }
4534
4535 #[test]
4536 fn test_prune_account_history_batch_multiple_sorted_targets() {
4537 let temp_dir = TempDir::new().unwrap();
4538 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4539
4540 let addr1 = Address::from([0x01; 20]);
4541 let addr2 = Address::from([0x02; 20]);
4542 let addr3 = Address::from([0x03; 20]);
4543
4544 let mut batch = provider.batch();
4546 batch
4547 .put::<tables::AccountsHistory>(
4548 ShardedKey::new(addr1, u64::MAX),
4549 &BlockNumberList::new_pre_sorted([10, 20, 30]),
4550 )
4551 .unwrap();
4552 batch
4553 .put::<tables::AccountsHistory>(
4554 ShardedKey::new(addr2, u64::MAX),
4555 &BlockNumberList::new_pre_sorted([5, 10, 15]),
4556 )
4557 .unwrap();
4558 batch
4559 .put::<tables::AccountsHistory>(
4560 ShardedKey::new(addr3, u64::MAX),
4561 &BlockNumberList::new_pre_sorted([100, 200]),
4562 )
4563 .unwrap();
4564 batch.commit().unwrap();
4565
4566 let mut targets = vec![(addr1, 15), (addr2, 10), (addr3, 50)];
4568 targets.sort_by_key(|(addr, _)| *addr);
4569
4570 let mut batch = provider.batch();
4571 let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4572 batch.commit().unwrap();
4573
4574 assert_eq!(outcomes.updated, 2);
4578 assert_eq!(outcomes.unchanged, 1);
4579
4580 let shards1 = provider.account_history_shards(addr1).unwrap();
4581 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
4582
4583 let shards2 = provider.account_history_shards(addr2).unwrap();
4584 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15]);
4585
4586 let shards3 = provider.account_history_shards(addr3).unwrap();
4587 assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![100, 200]);
4588 }
4589
4590 #[test]
4591 fn test_prune_account_history_batch_target_with_no_shards() {
4592 let temp_dir = TempDir::new().unwrap();
4593 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4594
4595 let addr1 = Address::from([0x01; 20]);
4596 let addr2 = Address::from([0x02; 20]); let addr3 = Address::from([0x03; 20]);
4598
4599 let mut batch = provider.batch();
4601 batch
4602 .put::<tables::AccountsHistory>(
4603 ShardedKey::new(addr1, u64::MAX),
4604 &BlockNumberList::new_pre_sorted([10, 20]),
4605 )
4606 .unwrap();
4607 batch
4608 .put::<tables::AccountsHistory>(
4609 ShardedKey::new(addr3, u64::MAX),
4610 &BlockNumberList::new_pre_sorted([30, 40]),
4611 )
4612 .unwrap();
4613 batch.commit().unwrap();
4614
4615 let mut targets = vec![(addr1, 15), (addr2, 100), (addr3, 35)];
4617 targets.sort_by_key(|(addr, _)| *addr);
4618
4619 let mut batch = provider.batch();
4620 let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4621 batch.commit().unwrap();
4622
4623 assert_eq!(outcomes.updated, 2);
4627 assert_eq!(outcomes.unchanged, 1);
4628
4629 let shards1 = provider.account_history_shards(addr1).unwrap();
4630 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20]);
4631
4632 let shards3 = provider.account_history_shards(addr3).unwrap();
4633 assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![40]);
4634 }
4635
4636 #[test]
4637 fn test_prune_storage_history_batch_multiple_sorted_targets() {
4638 let temp_dir = TempDir::new().unwrap();
4639 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4640
4641 let addr = Address::from([0x42; 20]);
4642 let slot1 = B256::from([0x01; 32]);
4643 let slot2 = B256::from([0x02; 32]);
4644
4645 let mut batch = provider.batch();
4647 batch
4648 .put::<tables::StoragesHistory>(
4649 StorageShardedKey::new(addr, slot1, u64::MAX),
4650 &BlockNumberList::new_pre_sorted([10, 20, 30]),
4651 )
4652 .unwrap();
4653 batch
4654 .put::<tables::StoragesHistory>(
4655 StorageShardedKey::new(addr, slot2, u64::MAX),
4656 &BlockNumberList::new_pre_sorted([5, 15, 25]),
4657 )
4658 .unwrap();
4659 batch.commit().unwrap();
4660
4661 let mut targets = vec![((addr, slot1), 15), ((addr, slot2), 10)];
4663 targets.sort_by_key(|((a, s), _)| (*a, *s));
4664
4665 let mut batch = provider.batch();
4666 let outcomes = batch.prune_storage_history_batch(&targets).unwrap();
4667 batch.commit().unwrap();
4668
4669 assert_eq!(outcomes.updated, 2);
4670
4671 let shards1 = provider.storage_history_shards(addr, slot1).unwrap();
4672 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
4673
4674 let shards2 = provider.storage_history_shards(addr, slot2).unwrap();
4675 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15, 25]);
4676 }
4677
4678 fn account_shard_layout(provider: &RocksDBProvider, address: Address) -> Vec<(u64, Vec<u64>)> {
4680 provider
4681 .account_history_shards(address)
4682 .unwrap()
4683 .into_iter()
4684 .map(|(key, list)| (key.highest_block_number, list.iter().collect::<Vec<_>>()))
4685 .collect()
4686 }
4687
4688 fn storage_shard_layout(
4690 provider: &RocksDBProvider,
4691 address: Address,
4692 storage_key: B256,
4693 ) -> Vec<(u64, Vec<u64>)> {
4694 provider
4695 .storage_history_shards(address, storage_key)
4696 .unwrap()
4697 .into_iter()
4698 .map(|(key, list)| {
4699 (key.sharded_key.highest_block_number, list.iter().collect::<Vec<_>>())
4700 })
4701 .collect()
4702 }
4703
4704 fn seed_three_storage_shards(provider: &RocksDBProvider, address: Address, storage_key: B256) {
4705 let mut batch = provider.batch();
4706 batch
4707 .put::<tables::StoragesHistory>(
4708 StorageShardedKey::new(address, storage_key, 100),
4709 &BlockNumberList::new_pre_sorted([10, 50, 100]),
4710 )
4711 .unwrap();
4712 batch
4713 .put::<tables::StoragesHistory>(
4714 StorageShardedKey::new(address, storage_key, 200),
4715 &BlockNumberList::new_pre_sorted([150, 200]),
4716 )
4717 .unwrap();
4718 batch
4719 .put::<tables::StoragesHistory>(
4720 StorageShardedKey::last(address, storage_key),
4721 &BlockNumberList::new_pre_sorted([250, 300]),
4722 )
4723 .unwrap();
4724 batch.commit().unwrap();
4725 }
4726
4727 #[test]
4728 fn test_prune_storage_history_batch_leaves_shards_above_target_untouched() {
4729 let temp_dir = TempDir::new().unwrap();
4730 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4731
4732 let addr = Address::from([0x42; 20]);
4733 let slot = B256::from([0x01; 32]);
4734 seed_three_storage_shards(&provider, addr, slot);
4735
4736 let mut batch = provider.batch();
4738 let outcomes = batch.prune_storage_history_batch(&[((addr, slot), 50)]).unwrap();
4739 batch.commit().unwrap();
4740
4741 assert_eq!(outcomes.updated, 1);
4742 assert_eq!(
4745 storage_shard_layout(&provider, addr, slot),
4746 vec![(100, vec![100]), (200, vec![150, 200]), (u64::MAX, vec![250, 300])]
4747 );
4748 }
4749
4750 #[test]
4751 fn test_prune_storage_history_batch_trims_sentinel_once_earlier_shards_expire() {
4752 let temp_dir = TempDir::new().unwrap();
4753 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4754
4755 let addr = Address::from([0x42; 20]);
4756 let slot = B256::from([0x01; 32]);
4757 seed_three_storage_shards(&provider, addr, slot);
4758
4759 let mut batch = provider.batch();
4761 let outcomes = batch.prune_storage_history_batch(&[((addr, slot), 250)]).unwrap();
4762 batch.commit().unwrap();
4763
4764 assert_eq!(outcomes.deleted, 1);
4765 assert_eq!(storage_shard_layout(&provider, addr, slot), vec![(u64::MAX, vec![300])]);
4766 }
4767
4768 #[test]
4769 fn test_prune_account_history_batch_leaves_shards_above_target_untouched() {
4770 let temp_dir = TempDir::new().unwrap();
4771 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4772
4773 let addr = Address::from([0x42; 20]);
4774
4775 let mut batch = provider.batch();
4776 batch
4777 .put::<tables::AccountsHistory>(
4778 ShardedKey::new(addr, 100),
4779 &BlockNumberList::new_pre_sorted([10, 50, 100]),
4780 )
4781 .unwrap();
4782 batch
4783 .put::<tables::AccountsHistory>(
4784 ShardedKey::new(addr, u64::MAX),
4785 &BlockNumberList::new_pre_sorted([250, 300]),
4786 )
4787 .unwrap();
4788 batch.commit().unwrap();
4789
4790 let mut batch = provider.batch();
4791 let outcomes = batch.prune_account_history_batch(&[(addr, 50)]).unwrap();
4792 batch.commit().unwrap();
4793
4794 assert_eq!(outcomes.updated, 1);
4795 assert_eq!(
4796 account_shard_layout(&provider, addr),
4797 vec![(100, vec![100]), (u64::MAX, vec![250, 300])]
4798 );
4799 }
4800
4801 #[test]
4802 fn test_prune_account_history_batch_seeks_after_stopping_early() {
4803 let temp_dir = TempDir::new().unwrap();
4804 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4805
4806 let addr1 = Address::from([0x01; 20]);
4807 let addr2 = Address::from([0x02; 20]);
4808
4809 let mut batch = provider.batch();
4810 batch
4811 .put::<tables::AccountsHistory>(
4812 ShardedKey::new(addr1, 100),
4813 &BlockNumberList::new_pre_sorted([10, 50, 100]),
4814 )
4815 .unwrap();
4816 batch
4817 .put::<tables::AccountsHistory>(
4818 ShardedKey::new(addr1, u64::MAX),
4819 &BlockNumberList::new_pre_sorted([250, 300]),
4820 )
4821 .unwrap();
4822 batch
4823 .put::<tables::AccountsHistory>(
4824 ShardedKey::new(addr2, u64::MAX),
4825 &BlockNumberList::new_pre_sorted([5, 10, 15]),
4826 )
4827 .unwrap();
4828 batch.commit().unwrap();
4829
4830 let mut batch = provider.batch();
4833 let outcomes = batch.prune_account_history_batch(&[(addr1, 50), (addr2, 10)]).unwrap();
4834 batch.commit().unwrap();
4835
4836 assert_eq!(outcomes.updated, 2);
4837 assert_eq!(
4838 account_shard_layout(&provider, addr1),
4839 vec![(100, vec![100]), (u64::MAX, vec![250, 300])]
4840 );
4841 assert_eq!(account_shard_layout(&provider, addr2), vec![(u64::MAX, vec![15])]);
4842 }
4843
4844 #[test]
4845 fn test_prune_storage_history_batch_seeks_after_stopping_early() {
4846 let temp_dir = TempDir::new().unwrap();
4847 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4848
4849 let addr = Address::from([0x42; 20]);
4850 let slot1 = B256::from([0x01; 32]);
4851 let slot2 = B256::from([0x02; 32]);
4852 seed_three_storage_shards(&provider, addr, slot1);
4853
4854 let mut batch = provider.batch();
4855 batch
4856 .put::<tables::StoragesHistory>(
4857 StorageShardedKey::last(addr, slot2),
4858 &BlockNumberList::new_pre_sorted([20, 40]),
4859 )
4860 .unwrap();
4861 batch.commit().unwrap();
4862
4863 let mut batch = provider.batch();
4866 let outcomes =
4867 batch.prune_storage_history_batch(&[((addr, slot1), 50), ((addr, slot2), 30)]).unwrap();
4868 batch.commit().unwrap();
4869
4870 assert_eq!(outcomes.updated, 2);
4871 assert_eq!(
4872 storage_shard_layout(&provider, addr, slot1),
4873 vec![(100, vec![100]), (200, vec![150, 200]), (u64::MAX, vec![250, 300])]
4874 );
4875 assert_eq!(storage_shard_layout(&provider, addr, slot2), vec![(u64::MAX, vec![40])]);
4876 }
4877}