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 DEFAULT_MAX_OPEN_FILES: i32 = 512;
126
127const DEFAULT_BYTES_PER_SYNC: u64 = 1_048_576;
129
130const DEFAULT_WRITE_BUFFER_SIZE: usize = 128 << 20;
136
137const DEFAULT_WRITE_BUFFER_MANAGER_SIZE: usize = 4 * 1024 * 1024 * 1024;
142
143const DEFAULT_COMPRESS_BUF_CAPACITY: usize = 4096;
147
148const DEFAULT_AUTO_COMMIT_THRESHOLD: usize = 512 * 1024 * 1024;
155
156const DEFAULT_BAL_MIN_BLOB_SIZE: u64 = 4 * 1024;
160
161const DEFAULT_BAL_BLOB_FILE_SIZE: u64 = 256 * 1024 * 1024;
163
164pub struct RocksDBBuilder {
166 path: PathBuf,
167 column_families: Vec<String>,
168 enable_metrics: bool,
169 enable_statistics: bool,
170 log_level: rocksdb::LogLevel,
171 block_cache: Cache,
172 read_only: bool,
173}
174
175impl fmt::Debug for RocksDBBuilder {
176 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
177 f.debug_struct("RocksDBBuilder")
178 .field("path", &self.path)
179 .field("column_families", &self.column_families)
180 .field("enable_metrics", &self.enable_metrics)
181 .finish()
182 }
183}
184
185impl RocksDBBuilder {
186 pub fn new(path: impl AsRef<Path>) -> Self {
188 let cache = Cache::new_lru_cache(DEFAULT_CACHE_SIZE);
189 Self {
190 path: path.as_ref().to_path_buf(),
191 column_families: Vec::new(),
192 enable_metrics: false,
193 enable_statistics: false,
194 log_level: rocksdb::LogLevel::Info,
195 block_cache: cache,
196 read_only: false,
197 }
198 }
199
200 fn default_table_options(cache: &Cache) -> BlockBasedOptions {
202 let mut table_options = BlockBasedOptions::default();
203 table_options.set_block_size(DEFAULT_BLOCK_SIZE);
204 table_options.set_cache_index_and_filter_blocks(true);
205 table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
206 table_options.set_block_cache(cache);
208 table_options
209 }
210
211 fn default_options(
213 log_level: rocksdb::LogLevel,
214 cache: &Cache,
215 enable_statistics: bool,
216 ) -> Options {
217 let table_options = Self::default_table_options(cache);
219
220 let mut options = Options::default();
221 options.set_block_based_table_factory(&table_options);
222 options.create_if_missing(true);
223 options.create_missing_column_families(true);
224 options.set_max_background_jobs(DEFAULT_MAX_BACKGROUND_JOBS);
225 options.set_bytes_per_sync(DEFAULT_BYTES_PER_SYNC);
226 let write_buffer_manager =
227 WriteBufferManager::new_write_buffer_manager(DEFAULT_WRITE_BUFFER_MANAGER_SIZE, true);
228 options.set_write_buffer_manager(&write_buffer_manager);
229
230 options.set_bottommost_compression_type(DBCompressionType::Zstd);
231 options.set_bottommost_zstd_max_train_bytes(0, true);
232 options.set_compression_type(DBCompressionType::Lz4);
233 options.set_compaction_pri(CompactionPri::MinOverlappingRatio);
234
235 options.set_log_level(log_level);
236
237 options.set_max_open_files(DEFAULT_MAX_OPEN_FILES);
238
239 options.set_wal_ttl_seconds(0);
242 options.set_wal_size_limit_mb(0);
243
244 if enable_statistics {
246 options.enable_statistics();
247 }
248
249 options
250 }
251
252 fn default_column_family_options(cache: &Cache) -> Options {
254 let table_options = Self::default_table_options(cache);
256
257 let mut cf_options = Options::default();
258 cf_options.set_block_based_table_factory(&table_options);
259 cf_options.set_level_compaction_dynamic_level_bytes(true);
260 cf_options.set_compression_type(DBCompressionType::Lz4);
262 cf_options.set_bottommost_compression_type(DBCompressionType::Zstd);
263 cf_options.set_bottommost_zstd_max_train_bytes(0, true);
265 cf_options.set_write_buffer_size(DEFAULT_WRITE_BUFFER_SIZE);
266
267 cf_options
268 }
269
270 fn block_access_lists_column_family_options(cache: &Cache) -> Options {
272 let mut cf_options = Self::default_column_family_options(cache);
273 cf_options.set_enable_blob_files(true);
274 cf_options.set_min_blob_size(DEFAULT_BAL_MIN_BLOB_SIZE);
275 cf_options.set_blob_file_size(DEFAULT_BAL_BLOB_FILE_SIZE);
276 cf_options.set_blob_compression_type(DBCompressionType::Lz4);
277 cf_options
278 }
279
280 fn tx_hash_numbers_column_family_options(cache: &Cache) -> Options {
287 let mut table_options = BlockBasedOptions::default();
288 table_options.set_block_size(DEFAULT_BLOCK_SIZE);
289 table_options.set_cache_index_and_filter_blocks(true);
290 table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
291 table_options.set_block_cache(cache);
292 let mut cf_options = Options::default();
296 cf_options.set_block_based_table_factory(&table_options);
297 cf_options.set_level_compaction_dynamic_level_bytes(true);
298 cf_options.set_compression_type(DBCompressionType::None);
301 cf_options.set_bottommost_compression_type(DBCompressionType::None);
302
303 cf_options
304 }
305
306 pub fn with_table<T: Table>(mut self) -> Self {
308 self.column_families.push(T::NAME.to_string());
309 self
310 }
311
312 pub fn with_default_tables(self) -> Self {
321 self.with_table::<tables::TransactionHashNumbers>()
322 .with_table::<tables::AccountsHistory>()
323 .with_table::<tables::StoragesHistory>()
324 .with_table::<tables::BlockAccessLists>()
325 .with_table::<tables::BlockAccessListBlockNumbers>()
326 }
327
328 pub const fn with_metrics(mut self) -> Self {
330 self.enable_metrics = true;
331 self
332 }
333
334 pub const fn with_statistics(mut self) -> Self {
336 self.enable_statistics = true;
337 self
338 }
339
340 pub const fn with_database_log_level(mut self, log_level: Option<LogLevel>) -> Self {
342 if let Some(level) = log_level {
343 self.log_level = convert_log_level(level);
344 }
345 self
346 }
347
348 pub fn with_block_cache_size(mut self, capacity_bytes: usize) -> Self {
350 self.block_cache = Cache::new_lru_cache(capacity_bytes);
351 self
352 }
353
354 pub const fn with_read_only(mut self, read_only: bool) -> Self {
362 self.read_only = read_only;
363 self
364 }
365
366 pub fn build(self) -> ProviderResult<RocksDBProvider> {
368 let options =
369 Self::default_options(self.log_level, &self.block_cache, self.enable_statistics);
370
371 let mut cf_descriptors: Vec<ColumnFamilyDescriptor> = self
372 .column_families
373 .iter()
374 .map(|name| {
375 let cf_options = if name == tables::TransactionHashNumbers::NAME {
376 Self::tx_hash_numbers_column_family_options(&self.block_cache)
377 } else if name == tables::BlockAccessLists::NAME {
378 Self::block_access_lists_column_family_options(&self.block_cache)
379 } else {
380 Self::default_column_family_options(&self.block_cache)
381 };
382 ColumnFamilyDescriptor::new(name.clone(), cf_options)
383 })
384 .collect();
385
386 if RocksDBProvider::exists(&self.path) {
389 let existing_column_families = DB::list_cf(&options, &self.path).map_err(|e| {
390 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
391 message: e.to_string().into(),
392 code: -1,
393 }))
394 })?;
395 let unknown_column_families: Vec<String> = existing_column_families
396 .into_iter()
397 .filter(|name| {
398 name != DEFAULT_COLUMN_FAMILY_NAME && !self.column_families.contains(name)
399 })
400 .collect();
401 if !unknown_column_families.is_empty() {
402 tracing::debug!(
403 target: "providers::rocksdb",
404 column_families = ?unknown_column_families,
405 "Preserving unknown column families"
406 );
407 cf_descriptors.extend(unknown_column_families.into_iter().map(|name| {
408 ColumnFamilyDescriptor::new(
409 name,
410 Self::default_column_family_options(&self.block_cache),
411 )
412 }));
413 }
414 }
415
416 let metrics = self.enable_metrics.then(RocksDBMetrics::default);
417
418 if self.read_only {
419 let mut options = options;
422 options.set_max_open_files(-1);
423
424 let secondary_path = self
425 .path
426 .parent()
427 .unwrap_or(&self.path)
428 .join(format!("rocksdb-secondary-tmp-{}", std::process::id()));
429 reth_fs_util::create_dir_all(&secondary_path).map_err(ProviderError::other)?;
430
431 let db = DB::open_cf_descriptors_as_secondary(
432 &options,
433 &self.path,
434 &secondary_path,
435 cf_descriptors,
436 )
437 .map_err(|e| {
438 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
439 message: e.to_string().into(),
440 code: -1,
441 }))
442 })?;
443 Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::Secondary {
444 db,
445 metrics,
446 secondary_path,
447 })))
448 } else {
449 let db =
454 OptimisticTransactionDB::open_cf_descriptors(&options, &self.path, cf_descriptors)
455 .map_err(|e| {
456 ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
457 message: e.to_string().into(),
458 code: -1,
459 }))
460 })?;
461 Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::ReadWrite { db, metrics })))
462 }
463 }
464}
465
466macro_rules! compress_to_buf_or_ref {
469 ($buf:expr, $value:expr) => {
470 if let Some(value) = $value.uncompressable_ref() {
471 Some(value)
472 } else {
473 $buf.clear();
474 $value.compress_to_buf(&mut $buf);
475 None
476 }
477 };
478}
479
480#[derive(Debug)]
482pub struct RocksDBProvider(Arc<RocksDBProviderInner>);
483
484enum RocksDBProviderInner {
486 ReadWrite {
488 db: OptimisticTransactionDB,
490 metrics: Option<RocksDBMetrics>,
492 },
493 Secondary {
497 db: DB,
499 metrics: Option<RocksDBMetrics>,
501 secondary_path: PathBuf,
503 },
504}
505
506impl RocksDBProviderInner {
507 const fn metrics(&self) -> Option<&RocksDBMetrics> {
509 match self {
510 Self::ReadWrite { metrics, .. } | Self::Secondary { metrics, .. } => metrics.as_ref(),
511 }
512 }
513
514 fn db_rw(&self) -> &OptimisticTransactionDB {
516 match self {
517 Self::ReadWrite { db, .. } => db,
518 Self::Secondary { .. } => {
519 panic!("Cannot perform write operation on secondary RocksDB provider")
520 }
521 }
522 }
523
524 fn cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
526 let cf = match self {
527 Self::ReadWrite { db, .. } => db.cf_handle(T::NAME),
528 Self::Secondary { db, .. } => db.cf_handle(T::NAME),
529 };
530 cf.ok_or_else(|| DatabaseError::Other(format!("Column family '{}' not found", T::NAME)))
531 }
532
533 fn get_cf(
535 &self,
536 cf: &rocksdb::ColumnFamily,
537 key: impl AsRef<[u8]>,
538 ) -> Result<Option<Vec<u8>>, rocksdb::Error> {
539 match self {
540 Self::ReadWrite { db, .. } => db.get_cf(cf, key),
541 Self::Secondary { db, .. } => db.get_cf(cf, key),
542 }
543 }
544
545 fn put_cf(
547 &self,
548 cf: &rocksdb::ColumnFamily,
549 key: impl AsRef<[u8]>,
550 value: impl AsRef<[u8]>,
551 ) -> Result<(), rocksdb::Error> {
552 self.db_rw().put_cf(cf, key, value)
553 }
554
555 fn delete_cf(
557 &self,
558 cf: &rocksdb::ColumnFamily,
559 key: impl AsRef<[u8]>,
560 ) -> Result<(), rocksdb::Error> {
561 self.db_rw().delete_cf(cf, key)
562 }
563
564 fn delete_range_cf<K: AsRef<[u8]>>(
566 &self,
567 cf: &rocksdb::ColumnFamily,
568 from: K,
569 to: K,
570 ) -> Result<(), rocksdb::Error> {
571 self.db_rw().delete_range_cf(cf, from, to)
572 }
573
574 fn iterator_cf(
576 &self,
577 cf: &rocksdb::ColumnFamily,
578 mode: IteratorMode<'_>,
579 ) -> RocksDBIterEnum<'_> {
580 match self {
581 Self::ReadWrite { db, .. } => RocksDBIterEnum::ReadWrite(db.iterator_cf(cf, mode)),
582 Self::Secondary { db, .. } => RocksDBIterEnum::ReadOnly(db.iterator_cf(cf, mode)),
583 }
584 }
585
586 fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
591 match self {
592 Self::ReadWrite { db, .. } => RocksDBRawIterEnum::ReadWrite(db.raw_iterator_cf(cf)),
593 Self::Secondary { db, .. } => RocksDBRawIterEnum::ReadOnly(db.raw_iterator_cf(cf)),
594 }
595 }
596
597 fn snapshot(&self) -> RocksReadSnapshotInner<'_> {
599 match self {
600 Self::ReadWrite { db, .. } => RocksReadSnapshotInner::ReadWrite(db.snapshot()),
601 Self::Secondary { db, .. } => RocksReadSnapshotInner::Secondary(db),
602 }
603 }
604
605 fn path(&self) -> &Path {
607 match self {
608 Self::ReadWrite { db, .. } => db.path(),
609 Self::Secondary { db, .. } => db.path(),
610 }
611 }
612
613 fn wal_size_bytes(&self) -> u64 {
617 let path = self.path();
618
619 match std::fs::read_dir(path) {
620 Ok(entries) => entries
621 .filter_map(|e| e.ok())
622 .filter(|e| e.path().extension().is_some_and(|ext| ext == "log"))
623 .filter_map(|e| e.metadata().ok())
624 .map(|m| m.len())
625 .sum(),
626 Err(_) => 0,
627 }
628 }
629
630 fn table_stats(&self) -> Vec<RocksDBTableStats> {
632 let mut stats = Vec::new();
633
634 macro_rules! collect_stats {
635 ($db:expr) => {
636 for cf_name in ROCKSDB_TABLES {
637 if let Some(cf) = $db.cf_handle(cf_name) {
638 let estimated_num_keys = $db
639 .property_int_value_cf(cf, rocksdb::properties::ESTIMATE_NUM_KEYS)
640 .ok()
641 .flatten()
642 .unwrap_or(0);
643
644 let sst_size = $db
646 .property_int_value_cf(cf, rocksdb::properties::LIVE_SST_FILES_SIZE)
647 .ok()
648 .flatten()
649 .unwrap_or(0);
650
651 let memtable_size = $db
652 .property_int_value_cf(cf, rocksdb::properties::SIZE_ALL_MEM_TABLES)
653 .ok()
654 .flatten()
655 .unwrap_or(0);
656
657 let estimated_size_bytes = sst_size + memtable_size;
658
659 let pending_compaction_bytes = $db
660 .property_int_value_cf(
661 cf,
662 rocksdb::properties::ESTIMATE_PENDING_COMPACTION_BYTES,
663 )
664 .ok()
665 .flatten()
666 .unwrap_or(0);
667
668 stats.push(RocksDBTableStats {
669 sst_size_bytes: sst_size,
670 memtable_size_bytes: memtable_size,
671 name: cf_name.to_string(),
672 estimated_num_keys,
673 estimated_size_bytes,
674 pending_compaction_bytes,
675 });
676 }
677 }
678 };
679 }
680
681 match self {
682 Self::ReadWrite { db, .. } => collect_stats!(db),
683 Self::Secondary { db, .. } => collect_stats!(db),
684 }
685
686 stats
687 }
688
689 fn db_stats(&self) -> RocksDBStats {
691 RocksDBStats { tables: self.table_stats(), wal_size_bytes: self.wal_size_bytes() }
692 }
693}
694
695impl fmt::Debug for RocksDBProviderInner {
696 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
697 match self {
698 Self::ReadWrite { metrics, .. } => f
699 .debug_struct("RocksDBProviderInner::ReadWrite")
700 .field("db", &"<OptimisticTransactionDB>")
701 .field("metrics", metrics)
702 .finish(),
703 Self::Secondary { metrics, .. } => f
704 .debug_struct("RocksDBProviderInner::Secondary")
705 .field("db", &"<DB (secondary)>")
706 .field("metrics", metrics)
707 .finish(),
708 }
709 }
710}
711
712impl Drop for RocksDBProviderInner {
713 fn drop(&mut self) {
714 match self {
715 Self::ReadWrite { db, .. } => {
716 if let Err(e) = db.flush_wal(true) {
719 tracing::warn!(target: "providers::rocksdb", ?e, "Failed to flush WAL on drop");
720 }
721 for cf_name in ROCKSDB_TABLES {
722 if let Some(cf) = db.cf_handle(cf_name) &&
723 let Err(e) = db.flush_cf(&cf)
724 {
725 tracing::warn!(target: "providers::rocksdb", cf = cf_name, ?e, "Failed to flush CF on drop");
726 }
727 }
728 db.cancel_all_background_work(true);
729 }
730 Self::Secondary { db, secondary_path, .. } => {
731 db.cancel_all_background_work(true);
732 let _ = std::fs::remove_dir_all(secondary_path);
733 }
734 }
735 }
736}
737
738impl Clone for RocksDBProvider {
739 fn clone(&self) -> Self {
740 Self(self.0.clone())
741 }
742}
743
744impl DatabaseMetrics for RocksDBProvider {
745 fn gauge_metrics(&self) -> Vec<(&'static str, f64, Vec<Label>)> {
746 let mut metrics = Vec::new();
747
748 for stat in self.table_stats() {
749 metrics.push((
750 "rocksdb.table_size",
751 stat.estimated_size_bytes as f64,
752 vec![Label::new("table", stat.name.clone())],
753 ));
754 metrics.push((
755 "rocksdb.table_entries",
756 stat.estimated_num_keys as f64,
757 vec![Label::new("table", stat.name.clone())],
758 ));
759 metrics.push((
760 "rocksdb.pending_compaction_bytes",
761 stat.pending_compaction_bytes as f64,
762 vec![Label::new("table", stat.name.clone())],
763 ));
764 metrics.push((
765 "rocksdb.sst_size",
766 stat.sst_size_bytes as f64,
767 vec![Label::new("table", stat.name.clone())],
768 ));
769 metrics.push((
770 "rocksdb.memtable_size",
771 stat.memtable_size_bytes as f64,
772 vec![Label::new("table", stat.name)],
773 ));
774 }
775
776 metrics.push(("rocksdb.wal_size", self.wal_size_bytes() as f64, vec![]));
778
779 metrics
780 }
781}
782
783impl RocksDBProvider {
784 pub fn new(path: impl AsRef<Path>) -> ProviderResult<Self> {
786 RocksDBBuilder::new(path).build()
787 }
788
789 pub fn builder(path: impl AsRef<Path>) -> RocksDBBuilder {
791 RocksDBBuilder::new(path)
792 }
793
794 pub fn exists(path: impl AsRef<Path>) -> bool {
799 path.as_ref().join("CURRENT").exists()
800 }
801
802 pub fn is_read_only(&self) -> bool {
804 matches!(self.0.as_ref(), RocksDBProviderInner::Secondary { .. })
805 }
806
807 pub fn try_catch_up_with_primary(&self) -> ProviderResult<()> {
812 match self.0.as_ref() {
813 RocksDBProviderInner::Secondary { db, .. } => {
814 db.try_catch_up_with_primary().map_err(|e| {
815 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
816 message: e.to_string().into(),
817 code: -1,
818 }))
819 })
820 }
821 _ => Ok(()),
822 }
823 }
824
825 pub fn snapshot(&self) -> RocksReadSnapshot<'_> {
829 RocksReadSnapshot { inner: self.0.snapshot(), provider: self }
830 }
831
832 pub fn tx(&self) -> RocksTx<'_> {
840 let write_options = synced_write_options();
841 let txn_options = OptimisticTransactionOptions::default();
842 let inner = self.0.db_rw().transaction_opt(&write_options, &txn_options);
843 RocksTx { inner, provider: self }
844 }
845
846 pub fn batch(&self) -> RocksDBBatch<'_> {
854 RocksDBBatch {
855 provider: self,
856 inner: WriteBatchWithTransaction::<true>::default(),
857 buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
858 auto_commit_threshold: None,
859 }
860 }
861
862 pub fn batch_with_auto_commit(&self) -> RocksDBBatch<'_> {
868 RocksDBBatch {
869 provider: self,
870 inner: WriteBatchWithTransaction::<true>::default(),
871 buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
872 auto_commit_threshold: Some(DEFAULT_AUTO_COMMIT_THRESHOLD),
873 }
874 }
875
876 fn get_cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
878 self.0.cf_handle::<T>()
879 }
880
881 fn execute_with_operation_metric<R>(
883 &self,
884 operation: RocksDBOperation,
885 table: &'static str,
886 f: impl FnOnce(&Self) -> R,
887 ) -> R {
888 let start = self.0.metrics().map(|_| Instant::now());
889 let res = f(self);
890
891 if let (Some(start), Some(metrics)) = (start, self.0.metrics()) {
892 metrics.record_operation(operation, table, start.elapsed());
893 }
894
895 res
896 }
897
898 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
900 self.get_encoded::<T>(&key.encode())
901 }
902
903 pub fn get_encoded<T: Table>(
905 &self,
906 key: &<T::Key as Encode>::Encoded,
907 ) -> ProviderResult<Option<T::Value>> {
908 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
909 let result = this.0.get_cf(this.get_cf_handle::<T>()?, key.as_ref()).map_err(|e| {
910 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
911 message: e.to_string().into(),
912 code: -1,
913 }))
914 })?;
915
916 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
917 })
918 }
919
920 pub fn get_raw<T: Table>(&self, key: T::Key) -> ProviderResult<Option<Vec<u8>>> {
922 let encoded = key.encode();
923 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
924 this.0.get_cf(this.get_cf_handle::<T>()?, encoded.as_ref()).map_err(|e| {
925 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
926 message: e.to_string().into(),
927 code: -1,
928 }))
929 })
930 })
931 }
932
933 pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
938 let encoded_key = key.encode();
939 self.put_encoded::<T>(&encoded_key, value)
940 }
941
942 pub fn put_encoded<T: Table>(
947 &self,
948 key: &<T::Key as Encode>::Encoded,
949 value: &T::Value,
950 ) -> ProviderResult<()> {
951 self.execute_with_operation_metric(RocksDBOperation::Put, T::NAME, |this| {
952 let mut buf = Vec::new();
956 let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
957
958 this.0.put_cf(this.get_cf_handle::<T>()?, key, value_bytes).map_err(|e| {
959 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
960 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
961 operation: DatabaseWriteOperation::PutUpsert,
962 table_name: T::NAME,
963 key: key.as_ref().to_vec(),
964 })))
965 })
966 })
967 }
968
969 pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
974 self.execute_with_operation_metric(RocksDBOperation::Delete, T::NAME, |this| {
975 this.0.delete_cf(this.get_cf_handle::<T>()?, key.encode().as_ref()).map_err(|e| {
976 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
977 message: e.to_string().into(),
978 code: -1,
979 }))
980 })
981 })
982 }
983
984 pub fn clear<T: Table>(&self) -> ProviderResult<()> {
990 let cf = self.get_cf_handle::<T>()?;
991
992 self.0.delete_range_cf(cf, &[] as &[u8], &[0xFF; 256]).map_err(|e| {
993 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
994 message: e.to_string().into(),
995 code: -1,
996 }))
997 })?;
998
999 Ok(())
1000 }
1001
1002 fn get_boundary<T: Table>(
1004 &self,
1005 mode: IteratorMode<'_>,
1006 ) -> ProviderResult<Option<(T::Key, T::Value)>> {
1007 self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
1008 let cf = this.get_cf_handle::<T>()?;
1009 let mut iter = this.0.iterator_cf(cf, mode);
1010
1011 match iter.next() {
1012 Some(Ok((key_bytes, value_bytes))) => {
1013 let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
1014 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1015 let value = T::Value::decompress(&value_bytes)
1016 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1017 Ok(Some((key, value)))
1018 }
1019 Some(Err(e)) => {
1020 Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1021 message: e.to_string().into(),
1022 code: -1,
1023 })))
1024 }
1025 None => Ok(None),
1026 }
1027 })
1028 }
1029
1030 #[inline]
1032 pub fn first<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1033 self.get_boundary::<T>(IteratorMode::Start)
1034 }
1035
1036 #[inline]
1038 pub fn last<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1039 self.get_boundary::<T>(IteratorMode::End)
1040 }
1041
1042 pub fn iter<T: Table>(&self) -> ProviderResult<RocksDBIter<'_, T>> {
1046 let cf = self.get_cf_handle::<T>()?;
1047 let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1048 Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1049 }
1050
1051 pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksDBIter<'_, T>> {
1055 let cf = self.get_cf_handle::<T>()?;
1056 let encoded_key = key.encode();
1057 let iter = self
1058 .0
1059 .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
1060 Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1061 }
1062
1063 pub fn table_stats(&self) -> Vec<RocksDBTableStats> {
1067 self.0.table_stats()
1068 }
1069
1070 pub fn wal_size_bytes(&self) -> u64 {
1076 self.0.wal_size_bytes()
1077 }
1078
1079 pub fn db_stats(&self) -> RocksDBStats {
1083 self.0.db_stats()
1084 }
1085
1086 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(tables = ?tables))]
1097 pub fn flush(&self, tables: &[&'static str]) -> ProviderResult<()> {
1098 let db = self.0.db_rw();
1099
1100 for cf_name in tables {
1101 if let Some(cf) = db.cf_handle(cf_name) {
1102 db.flush_cf(&cf).map_err(|e| {
1103 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1104 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1105 operation: DatabaseWriteOperation::Flush,
1106 table_name: cf_name,
1107 key: Vec::new(),
1108 })))
1109 })?;
1110 }
1111 }
1112
1113 db.flush_wal(true).map_err(|e| {
1114 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1115 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1116 operation: DatabaseWriteOperation::Flush,
1117 table_name: "WAL",
1118 key: Vec::new(),
1119 })))
1120 })?;
1121
1122 Ok(())
1123 }
1124
1125 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1137 pub fn flush_and_compact(&self) -> ProviderResult<()> {
1138 self.flush(ROCKSDB_TABLES)?;
1139
1140 let db = self.0.db_rw();
1141
1142 for cf_name in ROCKSDB_TABLES {
1143 if let Some(cf) = db.cf_handle(cf_name) {
1144 db.compact_range_cf(&cf, None::<&[u8]>, None::<&[u8]>);
1145 }
1146 }
1147
1148 Ok(())
1149 }
1150
1151 pub fn raw_iter<T: Table>(&self) -> ProviderResult<RocksDBRawIter<'_>> {
1155 let cf = self.get_cf_handle::<T>()?;
1156 let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1157 Ok(RocksDBRawIter { inner: iter })
1158 }
1159
1160 pub(crate) fn raw_key_iter_from<T: Table>(
1162 &self,
1163 key: T::Key,
1164 ) -> ProviderResult<RocksDBRawKeyIter<'_>> {
1165 let cf = self.get_cf_handle::<T>()?;
1166 let encoded_key = key.encode();
1167 let mut iter = self.0.raw_iterator_cf(cf);
1168 iter.seek(encoded_key.as_ref());
1169 Ok(RocksDBRawKeyIter { inner: iter })
1170 }
1171
1172 pub fn account_history_shards(
1177 &self,
1178 address: Address,
1179 ) -> ProviderResult<Vec<(ShardedKey<Address>, BlockNumberList)>> {
1180 let cf = self.get_cf_handle::<tables::AccountsHistory>()?;
1182
1183 let start_key = ShardedKey::new(address, 0u64);
1186 let start_bytes = start_key.encode();
1187
1188 let iter = self
1190 .0
1191 .iterator_cf(cf, IteratorMode::From(start_bytes.as_ref(), rocksdb::Direction::Forward));
1192
1193 let mut result = Vec::new();
1194 for item in iter {
1195 match item {
1196 Ok((key_bytes, value_bytes)) => {
1197 let key = ShardedKey::<Address>::decode(&key_bytes)
1199 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1200
1201 if key.key != address {
1203 break;
1204 }
1205
1206 let value = BlockNumberList::decompress(&value_bytes)
1208 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1209
1210 result.push((key, value));
1211 }
1212 Err(e) => {
1213 return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1214 message: e.to_string().into(),
1215 code: -1,
1216 })));
1217 }
1218 }
1219 }
1220
1221 Ok(result)
1222 }
1223
1224 pub fn storage_history_shards(
1229 &self,
1230 address: Address,
1231 storage_key: B256,
1232 ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1233 let cf = self.get_cf_handle::<tables::StoragesHistory>()?;
1234
1235 let start_key = StorageShardedKey::new(address, storage_key, 0u64);
1236 let start_bytes = start_key.encode();
1237
1238 let iter = self
1239 .0
1240 .iterator_cf(cf, IteratorMode::From(start_bytes.as_ref(), rocksdb::Direction::Forward));
1241
1242 let mut result = Vec::new();
1243 for item in iter {
1244 match item {
1245 Ok((key_bytes, value_bytes)) => {
1246 let key = StorageShardedKey::decode(&key_bytes)
1247 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1248
1249 if key.address != address || key.sharded_key.key != storage_key {
1250 break;
1251 }
1252
1253 let value = BlockNumberList::decompress(&value_bytes)
1254 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1255
1256 result.push((key, value));
1257 }
1258 Err(e) => {
1259 return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1260 message: e.to_string().into(),
1261 code: -1,
1262 })));
1263 }
1264 }
1265 }
1266
1267 Ok(result)
1268 }
1269
1270 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1278 pub fn unwind_account_history_indices(
1279 &self,
1280 last_indices: &[(Address, BlockNumber)],
1281 ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1282 let mut address_min_block: AddressMap<BlockNumber> =
1283 AddressMap::with_capacity_and_hasher(last_indices.len(), Default::default());
1284 for &(address, block_number) in last_indices {
1285 address_min_block
1286 .entry(address)
1287 .and_modify(|min| *min = (*min).min(block_number))
1288 .or_insert(block_number);
1289 }
1290
1291 let mut batch = self.batch();
1292 for (address, min_block) in address_min_block {
1293 match min_block.checked_sub(1) {
1294 Some(keep_to) => batch.unwind_account_history_to(address, keep_to)?,
1295 None => batch.clear_account_history(address)?,
1296 }
1297 }
1298
1299 Ok(batch.into_inner())
1300 }
1301
1302 pub fn unwind_storage_history_indices(
1310 &self,
1311 storage_changesets: &[(Address, B256, BlockNumber)],
1312 ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1313 let mut key_min_block: HashMap<(Address, B256), BlockNumber> =
1314 HashMap::with_capacity_and_hasher(storage_changesets.len(), Default::default());
1315 for &(address, storage_key, block_number) in storage_changesets {
1316 key_min_block
1317 .entry((address, storage_key))
1318 .and_modify(|min| *min = (*min).min(block_number))
1319 .or_insert(block_number);
1320 }
1321
1322 let mut batch = self.batch();
1323 for ((address, storage_key), min_block) in key_min_block {
1324 match min_block.checked_sub(1) {
1325 Some(keep_to) => batch.unwind_storage_history_to(address, storage_key, keep_to)?,
1326 None => batch.clear_storage_history(address, storage_key)?,
1327 }
1328 }
1329
1330 Ok(batch.into_inner())
1331 }
1332
1333 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1335 pub fn write_batch<F>(&self, f: F) -> ProviderResult<()>
1336 where
1337 F: FnOnce(&mut RocksDBBatch<'_>) -> ProviderResult<()>,
1338 {
1339 self.execute_with_operation_metric(RocksDBOperation::BatchWrite, "Batch", |this| {
1340 let mut batch_handle = this.batch();
1341 f(&mut batch_handle)?;
1342 batch_handle.commit()
1343 })
1344 }
1345
1346 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = batch.len(), batch_size = batch.size_in_bytes()))]
1354 pub fn commit_batch(&self, batch: WriteBatchWithTransaction<true>) -> ProviderResult<()> {
1355 self.0.db_rw().write_opt(batch, &synced_write_options()).map_err(|e| {
1356 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1357 message: e.to_string().into(),
1358 code: -1,
1359 }))
1360 })
1361 }
1362
1363 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(num_blocks = blocks.len(), first_block = ctx.first_block_number))]
1369 pub(crate) fn write_blocks_data<N: reth_node_types::NodePrimitives>(
1370 &self,
1371 blocks: &[ExecutedBlock<N>],
1372 tx_nums: &[TxNumber],
1373 ctx: RocksDBWriteCtx,
1374 runtime: &reth_tasks::Runtime,
1375 ) -> ProviderResult<()> {
1376 if !ctx.storage_settings.storage_v2 {
1377 return Ok(());
1378 }
1379
1380 let mut r_tx_hash = None;
1381 let mut r_account_history = None;
1382 let mut r_storage_history = None;
1383
1384 let write_tx_hash =
1385 ctx.storage_settings.storage_v2 && ctx.prune_tx_lookup.is_none_or(|m| !m.is_full());
1386 let write_account_history = ctx.storage_settings.storage_v2;
1387 let write_storage_history = ctx.storage_settings.storage_v2;
1388
1389 let span = tracing::Span::current();
1392 runtime.storage_pool().in_place_scope(|s| {
1393 if write_tx_hash {
1394 s.spawn(|_| {
1395 let _guard = span.enter();
1396 r_tx_hash = Some(self.write_tx_hash_numbers(blocks, tx_nums, &ctx));
1397 });
1398 }
1399
1400 if write_account_history {
1401 s.spawn(|_| {
1402 let _guard = span.enter();
1403 r_account_history = Some(self.write_account_history(blocks, &ctx));
1404 });
1405 }
1406
1407 if write_storage_history {
1408 s.spawn(|_| {
1409 let _guard = span.enter();
1410 r_storage_history = Some(self.write_storage_history(blocks, &ctx));
1411 });
1412 }
1413 });
1414
1415 if write_tx_hash {
1416 r_tx_hash.ok_or_else(|| {
1417 ProviderError::Database(DatabaseError::Other(
1418 "rocksdb tx-hash write thread panicked".into(),
1419 ))
1420 })??;
1421 }
1422 if write_account_history {
1423 r_account_history.ok_or_else(|| {
1424 ProviderError::Database(DatabaseError::Other(
1425 "rocksdb account-history write thread panicked".into(),
1426 ))
1427 })??;
1428 }
1429 if write_storage_history {
1430 r_storage_history.ok_or_else(|| {
1431 ProviderError::Database(DatabaseError::Other(
1432 "rocksdb storage-history write thread panicked".into(),
1433 ))
1434 })??;
1435 }
1436
1437 Ok(())
1438 }
1439
1440 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1442 fn write_tx_hash_numbers<N: reth_node_types::NodePrimitives>(
1443 &self,
1444 blocks: &[ExecutedBlock<N>],
1445 tx_nums: &[TxNumber],
1446 ctx: &RocksDBWriteCtx,
1447 ) -> ProviderResult<()> {
1448 let mut batch = self.batch();
1449 for (block, &first_tx_num) in blocks.iter().zip(tx_nums) {
1450 let body = block.recovered_block().body();
1451 for (tx_num, transaction) in (first_tx_num..).zip(body.transactions_iter()) {
1452 batch.put::<tables::TransactionHashNumbers>(*transaction.tx_hash(), &tx_num)?;
1453 }
1454 }
1455 ctx.pending_batches.lock().push(batch.into_inner());
1456 Ok(())
1457 }
1458
1459 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1463 fn write_account_history<N: reth_node_types::NodePrimitives>(
1464 &self,
1465 blocks: &[ExecutedBlock<N>],
1466 ctx: &RocksDBWriteCtx,
1467 ) -> ProviderResult<()> {
1468 let mut batch = self.batch();
1469 let mut account_history: BTreeMap<Address, Vec<u64>> = BTreeMap::new();
1470
1471 for (block_idx, block) in blocks.iter().enumerate() {
1472 let block_number = ctx.first_block_number + block_idx as u64;
1473 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1474
1475 for account_block_reverts in reverts.accounts {
1478 for (address, _) in account_block_reverts {
1479 account_history.entry(address).or_default().push(block_number);
1480 }
1481 }
1482 }
1483
1484 for (address, indices) in account_history {
1486 batch.append_account_history_shard(address, indices)?;
1487 }
1488 ctx.pending_batches.lock().push(batch.into_inner());
1489 Ok(())
1490 }
1491
1492 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1496 fn write_storage_history<N: reth_node_types::NodePrimitives>(
1497 &self,
1498 blocks: &[ExecutedBlock<N>],
1499 ctx: &RocksDBWriteCtx,
1500 ) -> ProviderResult<()> {
1501 let mut storage_history: BTreeMap<(Address, B256), Vec<u64>> = BTreeMap::new();
1502
1503 for (block_idx, block) in blocks.iter().enumerate() {
1504 let block_number = ctx.first_block_number + block_idx as u64;
1505 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1506
1507 for storage_block_reverts in reverts.storage {
1510 for revert in storage_block_reverts {
1511 for (slot, _) in revert.storage_revert {
1512 let plain_key = B256::new(slot.to_be_bytes());
1513 storage_history
1514 .entry((revert.address, plain_key))
1515 .or_default()
1516 .push(block_number);
1517 }
1518 }
1519 }
1520 }
1521
1522 let shard_puts = storage_history
1523 .into_par_iter()
1524 .map(|((address, slot), indices)| {
1525 self.storage_history_shards_to_put(address, slot, indices)
1526 })
1527 .collect::<ProviderResult<Vec<_>>>()?;
1528
1529 let mut batch = self.batch();
1530 for shards in shard_puts {
1531 for (key, shard) in shards {
1532 batch.put::<tables::StoragesHistory>(key, &shard)?;
1533 }
1534 }
1535 ctx.pending_batches.lock().push(batch.into_inner());
1536 Ok(())
1537 }
1538
1539 fn storage_history_shards_to_put(
1542 &self,
1543 address: Address,
1544 storage_key: B256,
1545 indices: Vec<u64>,
1546 ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1547 if indices.is_empty() {
1548 return Ok(Vec::new());
1549 }
1550
1551 debug_assert!(
1552 indices.windows(2).all(|w| w[0] < w[1]),
1553 "indices must be strictly increasing: {:?}",
1554 indices
1555 );
1556
1557 let last_key = StorageShardedKey::last(address, storage_key);
1558 let last_shard_opt = self.get::<tables::StoragesHistory>(last_key.clone())?;
1559 let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
1560
1561 last_shard.append(indices).map_err(ProviderError::other)?;
1562
1563 if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
1564 return Ok(vec![(last_key, last_shard)]);
1565 }
1566
1567 let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
1568 let mut chunks_peekable = chunks.into_iter().peekable();
1569 let mut shards = Vec::new();
1570
1571 while let Some(chunk) = chunks_peekable.next() {
1572 let shard = BlockNumberList::new_pre_sorted(chunk);
1573 let highest_block_number = if chunks_peekable.peek().is_some() {
1574 shard.iter().next_back().expect("`chunks` does not return empty list")
1575 } else {
1576 u64::MAX
1577 };
1578
1579 shards
1580 .push((StorageShardedKey::new(address, storage_key, highest_block_number), shard));
1581 }
1582
1583 Ok(shards)
1584 }
1585}
1586
1587pub struct RocksReadSnapshot<'db> {
1595 inner: RocksReadSnapshotInner<'db>,
1596 provider: &'db RocksDBProvider,
1597}
1598
1599enum RocksReadSnapshotInner<'db> {
1601 ReadWrite(SnapshotWithThreadMode<'db, OptimisticTransactionDB>),
1603 Secondary(&'db DB),
1605}
1606
1607impl<'db> RocksReadSnapshotInner<'db> {
1608 fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
1610 match self {
1611 Self::ReadWrite(snap) => RocksDBRawIterEnum::ReadWrite(snap.raw_iterator_cf(cf)),
1612 Self::Secondary(db) => RocksDBRawIterEnum::ReadOnly(db.raw_iterator_cf(cf)),
1613 }
1614 }
1615}
1616
1617impl fmt::Debug for RocksReadSnapshot<'_> {
1618 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1619 f.debug_struct("RocksReadSnapshot")
1620 .field("provider", &self.provider)
1621 .finish_non_exhaustive()
1622 }
1623}
1624
1625impl<'db> RocksReadSnapshot<'db> {
1626 fn cf_handle<T: Table>(&self) -> Result<&'db rocksdb::ColumnFamily, DatabaseError> {
1628 self.provider.get_cf_handle::<T>()
1629 }
1630
1631 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
1633 let encoded_key = key.encode();
1634 let cf = self.cf_handle::<T>()?;
1635 let result = match &self.inner {
1636 RocksReadSnapshotInner::ReadWrite(snap) => snap.get_cf(cf, encoded_key.as_ref()),
1637 RocksReadSnapshotInner::Secondary(db) => db.get_cf(cf, encoded_key.as_ref()),
1638 }
1639 .map_err(|e| {
1640 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1641 message: e.to_string().into(),
1642 code: -1,
1643 }))
1644 })?;
1645
1646 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
1647 }
1648
1649 pub fn account_history_info(
1654 &self,
1655 address: Address,
1656 block_number: BlockNumber,
1657 lowest_available_block_number: Option<BlockNumber>,
1658 visible_tip: BlockNumber,
1659 ) -> ProviderResult<HistoryInfo> {
1660 let key = ShardedKey::new(address, block_number);
1661 self.history_info::<tables::AccountsHistory>(
1662 key.encode().as_ref(),
1663 block_number,
1664 lowest_available_block_number,
1665 visible_tip,
1666 |key_bytes| Ok(<ShardedKey<Address> as Decode>::decode(key_bytes)?.key == address),
1667 |prev_bytes| {
1668 <ShardedKey<Address> as Decode>::decode(prev_bytes)
1669 .map(|k| k.key == address)
1670 .unwrap_or(false)
1671 },
1672 )
1673 }
1674
1675 pub fn storage_history_info(
1680 &self,
1681 address: Address,
1682 storage_key: B256,
1683 block_number: BlockNumber,
1684 lowest_available_block_number: Option<BlockNumber>,
1685 visible_tip: BlockNumber,
1686 ) -> ProviderResult<HistoryInfo> {
1687 let key = StorageShardedKey::new(address, storage_key, block_number);
1688 self.history_info::<tables::StoragesHistory>(
1689 key.encode().as_ref(),
1690 block_number,
1691 lowest_available_block_number,
1692 visible_tip,
1693 |key_bytes| {
1694 let k = <StorageShardedKey as Decode>::decode(key_bytes)?;
1695 Ok(k.address == address && k.sharded_key.key == storage_key)
1696 },
1697 |prev_bytes| {
1698 <StorageShardedKey as Decode>::decode(prev_bytes)
1699 .map(|k| k.address == address && k.sharded_key.key == storage_key)
1700 .unwrap_or(false)
1701 },
1702 )
1703 }
1704
1705 fn history_info<T>(
1711 &self,
1712 encoded_key: &[u8],
1713 block_number: BlockNumber,
1714 lowest_available_block_number: Option<BlockNumber>,
1715 visible_tip: BlockNumber,
1716 key_matches: impl FnOnce(&[u8]) -> Result<bool, reth_db_api::DatabaseError>,
1717 prev_key_matches: impl Fn(&[u8]) -> bool,
1718 ) -> ProviderResult<HistoryInfo>
1719 where
1720 T: Table<Value = BlockNumberList>,
1721 {
1722 let is_maybe_pruned = lowest_available_block_number.is_some();
1723 let fallback = || {
1724 Ok(if is_maybe_pruned {
1725 HistoryInfo::MaybeInPlainState
1726 } else {
1727 HistoryInfo::NotYetWritten
1728 })
1729 };
1730
1731 let cf = self.cf_handle::<T>()?;
1732 let mut iter = self.inner.raw_iterator_cf(cf);
1733
1734 iter.seek(encoded_key);
1735 iter.status().map_err(|e| {
1736 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1737 message: e.to_string().into(),
1738 code: -1,
1739 }))
1740 })?;
1741
1742 if !iter.valid() {
1743 return fallback();
1744 }
1745
1746 let Some(key_bytes) = iter.key() else {
1747 return fallback();
1748 };
1749 if !key_matches(key_bytes)? {
1750 return fallback();
1751 }
1752
1753 let Some(value_bytes) = iter.value() else {
1754 return fallback();
1755 };
1756 let chunk = BlockNumberList::decompress(value_bytes)?;
1757
1758 let (rank, found_block) = compute_history_rank(&chunk, block_number);
1759 let found_block = found_block.filter(|block| *block <= visible_tip);
1761
1762 let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) {
1763 iter.prev();
1764 iter.status().map_err(|e| {
1765 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1766 message: e.to_string().into(),
1767 code: -1,
1768 }))
1769 })?;
1770 let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches);
1771
1772 if found_block.is_none() && !has_prev {
1776 return fallback()
1777 }
1778
1779 !has_prev
1780 } else {
1781 false
1782 };
1783
1784 Ok(HistoryInfo::from_lookup(
1785 found_block,
1786 is_before_first_write,
1787 lowest_available_block_number,
1788 ))
1789 }
1790}
1791
1792#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1794pub enum PruneShardOutcome {
1795 Deleted,
1797 Updated,
1799 Unchanged,
1801}
1802
1803#[derive(Debug, Default, Clone, Copy)]
1805pub struct PrunedIndices {
1806 pub deleted: usize,
1808 pub updated: usize,
1810 pub unchanged: usize,
1812}
1813
1814#[must_use = "batch must be committed"]
1824pub struct RocksDBBatch<'a> {
1825 provider: &'a RocksDBProvider,
1826 inner: WriteBatchWithTransaction<true>,
1827 buf: Vec<u8>,
1828 auto_commit_threshold: Option<usize>,
1830}
1831
1832impl fmt::Debug for RocksDBBatch<'_> {
1833 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1834 f.debug_struct("RocksDBBatch")
1835 .field("provider", &self.provider)
1836 .field("batch", &"<WriteBatchWithTransaction>")
1837 .field("length", &self.inner.len())
1839 .field("size_in_bytes", &self.inner.size_in_bytes())
1842 .finish()
1843 }
1844}
1845
1846impl<'a> RocksDBBatch<'a> {
1847 pub fn put<T: Table>(&mut self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
1851 let encoded_key = key.encode();
1852 self.put_encoded::<T>(&encoded_key, value)
1853 }
1854
1855 pub fn put_encoded<T: Table>(
1859 &mut self,
1860 key: &<T::Key as Encode>::Encoded,
1861 value: &T::Value,
1862 ) -> ProviderResult<()> {
1863 let value_bytes = compress_to_buf_or_ref!(self.buf, value).unwrap_or(&self.buf);
1864 self.inner.put_cf(self.provider.get_cf_handle::<T>()?, key, value_bytes);
1865 self.maybe_auto_commit()?;
1866 Ok(())
1867 }
1868
1869 pub fn delete<T: Table>(&mut self, key: T::Key) -> ProviderResult<()> {
1873 self.inner.delete_cf(self.provider.get_cf_handle::<T>()?, key.encode().as_ref());
1874 self.maybe_auto_commit()?;
1875 Ok(())
1876 }
1877
1878 fn maybe_auto_commit(&mut self) -> ProviderResult<()> {
1883 if let Some(threshold) = self.auto_commit_threshold &&
1884 self.inner.size_in_bytes() >= threshold
1885 {
1886 tracing::debug!(
1887 target: "providers::rocksdb",
1888 batch_size = self.inner.size_in_bytes(),
1889 threshold,
1890 "Auto-committing RocksDB batch"
1891 );
1892 let old_batch = std::mem::take(&mut self.inner);
1893 self.provider.0.db_rw().write_opt(old_batch, &synced_write_options()).map_err(|e| {
1894 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1895 message: e.to_string().into(),
1896 code: -1,
1897 }))
1898 })?;
1899 }
1900 Ok(())
1901 }
1902
1903 #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = self.inner.len(), batch_size = self.inner.size_in_bytes()))]
1910 pub fn commit(self) -> ProviderResult<()> {
1911 self.provider.0.db_rw().write_opt(self.inner, &synced_write_options()).map_err(|e| {
1912 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1913 message: e.to_string().into(),
1914 code: -1,
1915 }))
1916 })
1917 }
1918
1919 pub fn len(&self) -> usize {
1921 self.inner.len()
1922 }
1923
1924 pub fn is_empty(&self) -> bool {
1926 self.inner.is_empty()
1927 }
1928
1929 pub fn size_in_bytes(&self) -> usize {
1931 self.inner.size_in_bytes()
1932 }
1933
1934 pub const fn provider(&self) -> &RocksDBProvider {
1936 self.provider
1937 }
1938
1939 pub fn into_inner(self) -> WriteBatchWithTransaction<true> {
1943 self.inner
1944 }
1945
1946 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
1951 self.provider.get::<T>(key)
1952 }
1953
1954 pub fn append_account_history_shard(
1966 &mut self,
1967 address: Address,
1968 indices: impl IntoIterator<Item = u64>,
1969 ) -> ProviderResult<()> {
1970 let indices: Vec<u64> = indices.into_iter().collect();
1971
1972 if indices.is_empty() {
1973 return Ok(());
1974 }
1975
1976 debug_assert!(
1977 indices.windows(2).all(|w| w[0] < w[1]),
1978 "indices must be strictly increasing: {:?}",
1979 indices
1980 );
1981
1982 let last_key = ShardedKey::new(address, u64::MAX);
1983 let last_shard_opt = self.provider.get::<tables::AccountsHistory>(last_key.clone())?;
1984 let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
1985
1986 last_shard.append(indices).map_err(ProviderError::other)?;
1987
1988 if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
1990 self.put::<tables::AccountsHistory>(last_key, &last_shard)?;
1991 return Ok(());
1992 }
1993
1994 let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
1996 let mut chunks_peekable = chunks.into_iter().peekable();
1997
1998 while let Some(chunk) = chunks_peekable.next() {
1999 let shard = BlockNumberList::new_pre_sorted(chunk);
2000 let highest_block_number = if chunks_peekable.peek().is_some() {
2001 shard.iter().next_back().expect("`chunks` does not return empty list")
2002 } else {
2003 u64::MAX
2004 };
2005
2006 self.put::<tables::AccountsHistory>(
2007 ShardedKey::new(address, highest_block_number),
2008 &shard,
2009 )?;
2010 }
2011
2012 Ok(())
2013 }
2014
2015 pub fn append_storage_history_shard(
2027 &mut self,
2028 address: Address,
2029 storage_key: B256,
2030 indices: impl IntoIterator<Item = u64>,
2031 ) -> ProviderResult<()> {
2032 let indices: Vec<u64> = indices.into_iter().collect();
2033
2034 for (key, shard) in
2035 self.provider.storage_history_shards_to_put(address, storage_key, indices)?
2036 {
2037 self.put::<tables::StoragesHistory>(key, &shard)?;
2038 }
2039
2040 Ok(())
2041 }
2042
2043 pub fn unwind_account_history_to(
2050 &mut self,
2051 address: Address,
2052 keep_to: BlockNumber,
2053 ) -> ProviderResult<()> {
2054 let shards = self.provider.account_history_shards(address)?;
2055 if shards.is_empty() {
2056 return Ok(());
2057 }
2058
2059 let boundary_idx = shards.iter().position(|(key, _)| {
2062 key.highest_block_number == u64::MAX || key.highest_block_number > keep_to
2063 });
2064
2065 let Some(boundary_idx) = boundary_idx else {
2067 let (last_key, last_value) = shards.last().expect("shards is non-empty");
2068 if last_key.highest_block_number != u64::MAX {
2069 self.delete::<tables::AccountsHistory>(last_key.clone())?;
2070 self.put::<tables::AccountsHistory>(
2071 ShardedKey::new(address, u64::MAX),
2072 last_value,
2073 )?;
2074 }
2075 return Ok(());
2076 };
2077
2078 for (key, _) in shards.iter().skip(boundary_idx + 1) {
2080 self.delete::<tables::AccountsHistory>(key.clone())?;
2081 }
2082
2083 let (boundary_key, boundary_list) = &shards[boundary_idx];
2085
2086 self.delete::<tables::AccountsHistory>(boundary_key.clone())?;
2088
2089 let new_last =
2091 BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2092
2093 if new_last.is_empty() {
2094 if boundary_idx == 0 {
2097 return Ok(());
2099 }
2100
2101 let (prev_key, prev_value) = &shards[boundary_idx - 1];
2102 if prev_key.highest_block_number != u64::MAX {
2103 self.delete::<tables::AccountsHistory>(prev_key.clone())?;
2104 self.put::<tables::AccountsHistory>(
2105 ShardedKey::new(address, u64::MAX),
2106 prev_value,
2107 )?;
2108 }
2109 return Ok(());
2110 }
2111
2112 self.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &new_last)?;
2113
2114 Ok(())
2115 }
2116
2117 #[expect(clippy::too_many_arguments)]
2123 fn prune_history_shards_inner<K>(
2124 &mut self,
2125 shards: Vec<(K, BlockNumberList)>,
2126 to_block: BlockNumber,
2127 get_highest: impl Fn(&K) -> u64,
2128 is_sentinel: impl Fn(&K) -> bool,
2129 delete_shard: impl Fn(&mut Self, K) -> ProviderResult<()>,
2130 put_shard: impl Fn(&mut Self, K, &BlockNumberList) -> ProviderResult<()>,
2131 create_sentinel: impl Fn() -> K,
2132 ) -> ProviderResult<PruneShardOutcome>
2133 where
2134 K: Clone,
2135 {
2136 if shards.is_empty() {
2137 return Ok(PruneShardOutcome::Unchanged);
2138 }
2139
2140 let mut deleted = false;
2141 let mut updated = false;
2142 let mut last_remaining: Option<(K, BlockNumberList)> = None;
2143
2144 for (key, block_list) in shards {
2145 if !is_sentinel(&key) && get_highest(&key) <= to_block {
2146 delete_shard(self, key)?;
2147 deleted = true;
2148 } else {
2149 let original_len = block_list.len();
2150 let filtered =
2151 BlockNumberList::new_pre_sorted(block_list.iter().filter(|&b| b > to_block));
2152
2153 if filtered.is_empty() {
2154 delete_shard(self, key)?;
2155 deleted = true;
2156 } else if filtered.len() < original_len {
2157 put_shard(self, key.clone(), &filtered)?;
2158 last_remaining = Some((key, filtered));
2159 updated = true;
2160 } else {
2161 last_remaining = Some((key, block_list));
2162 }
2163 }
2164 }
2165
2166 if let Some((last_key, last_value)) = last_remaining &&
2167 !is_sentinel(&last_key)
2168 {
2169 delete_shard(self, last_key)?;
2170 put_shard(self, create_sentinel(), &last_value)?;
2171 updated = true;
2172 }
2173
2174 if deleted {
2175 Ok(PruneShardOutcome::Deleted)
2176 } else if updated {
2177 Ok(PruneShardOutcome::Updated)
2178 } else {
2179 Ok(PruneShardOutcome::Unchanged)
2180 }
2181 }
2182
2183 pub fn prune_account_history_to(
2188 &mut self,
2189 address: Address,
2190 to_block: BlockNumber,
2191 ) -> ProviderResult<PruneShardOutcome> {
2192 let shards = self.provider.account_history_shards(address)?;
2193 self.prune_history_shards_inner(
2194 shards,
2195 to_block,
2196 |key| key.highest_block_number,
2197 |key| key.highest_block_number == u64::MAX,
2198 |batch, key| batch.delete::<tables::AccountsHistory>(key),
2199 |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2200 || ShardedKey::new(address, u64::MAX),
2201 )
2202 }
2203
2204 pub fn prune_account_history_batch(
2213 &mut self,
2214 targets: &[(Address, BlockNumber)],
2215 ) -> ProviderResult<PrunedIndices> {
2216 if targets.is_empty() {
2217 return Ok(PrunedIndices::default());
2218 }
2219
2220 debug_assert!(
2221 targets.windows(2).all(|w| w[0].0 <= w[1].0),
2222 "prune_account_history_batch: targets must be sorted by address"
2223 );
2224
2225 const PREFIX_LEN: usize = 20;
2228
2229 let cf = self.provider.get_cf_handle::<tables::AccountsHistory>()?;
2230 let mut iter = self.provider.0.raw_iterator_cf(cf);
2231 let mut outcomes = PrunedIndices::default();
2232
2233 for (address, to_block) in targets {
2234 let start_key = ShardedKey::new(*address, 0u64).encode();
2236 let target_prefix = &start_key[..PREFIX_LEN];
2237
2238 let needs_seek = if iter.valid() {
2244 if let Some(current_key) = iter.key() {
2245 current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2249 } else {
2250 true
2251 }
2252 } else {
2253 true
2254 };
2255
2256 if needs_seek {
2257 iter.seek(start_key);
2258 iter.status().map_err(|e| {
2259 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2260 message: e.to_string().into(),
2261 code: -1,
2262 }))
2263 })?;
2264 }
2265
2266 let mut shards = Vec::new();
2268 while iter.valid() {
2269 let Some(key_bytes) = iter.key() else { break };
2270
2271 let current_prefix = key_bytes.get(..PREFIX_LEN);
2273 if current_prefix != Some(target_prefix) {
2274 break;
2275 }
2276
2277 let key = ShardedKey::<Address>::decode(key_bytes)
2279 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2280
2281 let Some(value_bytes) = iter.value() else { break };
2282 let value = BlockNumberList::decompress(value_bytes)
2283 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2284
2285 shards.push((key, value));
2286 iter.next();
2287 }
2288
2289 match self.prune_history_shards_inner(
2290 shards,
2291 *to_block,
2292 |key| key.highest_block_number,
2293 |key| key.highest_block_number == u64::MAX,
2294 |batch, key| batch.delete::<tables::AccountsHistory>(key),
2295 |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2296 || ShardedKey::new(*address, u64::MAX),
2297 )? {
2298 PruneShardOutcome::Deleted => outcomes.deleted += 1,
2299 PruneShardOutcome::Updated => outcomes.updated += 1,
2300 PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2301 }
2302 }
2303
2304 Ok(outcomes)
2305 }
2306
2307 pub fn prune_storage_history_to(
2313 &mut self,
2314 address: Address,
2315 storage_key: B256,
2316 to_block: BlockNumber,
2317 ) -> ProviderResult<PruneShardOutcome> {
2318 let shards = self.provider.storage_history_shards(address, storage_key)?;
2319 self.prune_history_shards_inner(
2320 shards,
2321 to_block,
2322 |key| key.sharded_key.highest_block_number,
2323 |key| key.sharded_key.highest_block_number == u64::MAX,
2324 |batch, key| batch.delete::<tables::StoragesHistory>(key),
2325 |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2326 || StorageShardedKey::last(address, storage_key),
2327 )
2328 }
2329
2330 pub fn prune_storage_history_batch(
2340 &mut self,
2341 targets: &[((Address, B256), BlockNumber)],
2342 ) -> ProviderResult<PrunedIndices> {
2343 if targets.is_empty() {
2344 return Ok(PrunedIndices::default());
2345 }
2346
2347 debug_assert!(
2348 targets.windows(2).all(|w| w[0].0 <= w[1].0),
2349 "prune_storage_history_batch: targets must be sorted by (address, storage_key)"
2350 );
2351
2352 const PREFIX_LEN: usize = 52;
2355
2356 let cf = self.provider.get_cf_handle::<tables::StoragesHistory>()?;
2357 let mut iter = self.provider.0.raw_iterator_cf(cf);
2358 let mut outcomes = PrunedIndices::default();
2359
2360 for ((address, storage_key), to_block) in targets {
2361 let start_key = StorageShardedKey::new(*address, *storage_key, 0u64).encode();
2363 let target_prefix = &start_key[..PREFIX_LEN];
2364
2365 let needs_seek = if iter.valid() {
2371 if let Some(current_key) = iter.key() {
2372 current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2376 } else {
2377 true
2378 }
2379 } else {
2380 true
2381 };
2382
2383 if needs_seek {
2384 iter.seek(start_key);
2385 iter.status().map_err(|e| {
2386 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2387 message: e.to_string().into(),
2388 code: -1,
2389 }))
2390 })?;
2391 }
2392
2393 let mut shards = Vec::new();
2395 while iter.valid() {
2396 let Some(key_bytes) = iter.key() else { break };
2397
2398 let current_prefix = key_bytes.get(..PREFIX_LEN);
2400 if current_prefix != Some(target_prefix) {
2401 break;
2402 }
2403
2404 let key = StorageShardedKey::decode(key_bytes)
2406 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2407
2408 let Some(value_bytes) = iter.value() else { break };
2409 let value = BlockNumberList::decompress(value_bytes)
2410 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2411
2412 shards.push((key, value));
2413 iter.next();
2414 }
2415
2416 match self.prune_history_shards_inner(
2418 shards,
2419 *to_block,
2420 |key| key.sharded_key.highest_block_number,
2421 |key| key.sharded_key.highest_block_number == u64::MAX,
2422 |batch, key| batch.delete::<tables::StoragesHistory>(key),
2423 |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2424 || StorageShardedKey::last(*address, *storage_key),
2425 )? {
2426 PruneShardOutcome::Deleted => outcomes.deleted += 1,
2427 PruneShardOutcome::Updated => outcomes.updated += 1,
2428 PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2429 }
2430 }
2431
2432 Ok(outcomes)
2433 }
2434
2435 pub fn unwind_storage_history_to(
2444 &mut self,
2445 address: Address,
2446 storage_key: B256,
2447 keep_to: BlockNumber,
2448 ) -> ProviderResult<()> {
2449 let shards = self.provider.storage_history_shards(address, storage_key)?;
2450 if shards.is_empty() {
2451 return Ok(());
2452 }
2453
2454 let boundary_idx = shards.iter().position(|(key, _)| {
2457 key.sharded_key.highest_block_number == u64::MAX ||
2458 key.sharded_key.highest_block_number > keep_to
2459 });
2460
2461 let Some(boundary_idx) = boundary_idx else {
2463 let (last_key, last_value) = shards.last().expect("shards is non-empty");
2464 if last_key.sharded_key.highest_block_number != u64::MAX {
2465 self.delete::<tables::StoragesHistory>(last_key.clone())?;
2466 self.put::<tables::StoragesHistory>(
2467 StorageShardedKey::last(address, storage_key),
2468 last_value,
2469 )?;
2470 }
2471 return Ok(());
2472 };
2473
2474 for (key, _) in shards.iter().skip(boundary_idx + 1) {
2476 self.delete::<tables::StoragesHistory>(key.clone())?;
2477 }
2478
2479 let (boundary_key, boundary_list) = &shards[boundary_idx];
2481
2482 self.delete::<tables::StoragesHistory>(boundary_key.clone())?;
2484
2485 let new_last =
2487 BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2488
2489 if new_last.is_empty() {
2490 if boundary_idx == 0 {
2493 return Ok(());
2495 }
2496
2497 let (prev_key, prev_value) = &shards[boundary_idx - 1];
2498 if prev_key.sharded_key.highest_block_number != u64::MAX {
2499 self.delete::<tables::StoragesHistory>(prev_key.clone())?;
2500 self.put::<tables::StoragesHistory>(
2501 StorageShardedKey::last(address, storage_key),
2502 prev_value,
2503 )?;
2504 }
2505 return Ok(());
2506 }
2507
2508 self.put::<tables::StoragesHistory>(
2509 StorageShardedKey::last(address, storage_key),
2510 &new_last,
2511 )?;
2512
2513 Ok(())
2514 }
2515
2516 pub fn clear_account_history(&mut self, address: Address) -> ProviderResult<()> {
2520 let shards = self.provider.account_history_shards(address)?;
2521 for (key, _) in shards {
2522 self.delete::<tables::AccountsHistory>(key)?;
2523 }
2524 Ok(())
2525 }
2526
2527 pub fn clear_storage_history(
2531 &mut self,
2532 address: Address,
2533 storage_key: B256,
2534 ) -> ProviderResult<()> {
2535 let shards = self.provider.storage_history_shards(address, storage_key)?;
2536 for (key, _) in shards {
2537 self.delete::<tables::StoragesHistory>(key)?;
2538 }
2539 Ok(())
2540 }
2541}
2542
2543pub struct RocksTx<'db> {
2553 inner: Transaction<'db, OptimisticTransactionDB>,
2554 provider: &'db RocksDBProvider,
2555}
2556
2557impl fmt::Debug for RocksTx<'_> {
2558 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2559 f.debug_struct("RocksTx").field("provider", &self.provider).finish_non_exhaustive()
2560 }
2561}
2562
2563impl<'db> RocksTx<'db> {
2564 pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
2566 let encoded_key = key.encode();
2567 self.get_encoded::<T>(&encoded_key)
2568 }
2569
2570 pub fn get_encoded<T: Table>(
2572 &self,
2573 key: &<T::Key as Encode>::Encoded,
2574 ) -> ProviderResult<Option<T::Value>> {
2575 let cf = self.provider.get_cf_handle::<T>()?;
2576 let result = self.inner.get_cf(cf, key.as_ref()).map_err(|e| {
2577 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2578 message: e.to_string().into(),
2579 code: -1,
2580 }))
2581 })?;
2582
2583 Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
2584 }
2585
2586 pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
2588 let encoded_key = key.encode();
2589 self.put_encoded::<T>(&encoded_key, value)
2590 }
2591
2592 pub fn put_encoded<T: Table>(
2594 &self,
2595 key: &<T::Key as Encode>::Encoded,
2596 value: &T::Value,
2597 ) -> ProviderResult<()> {
2598 let cf = self.provider.get_cf_handle::<T>()?;
2599 let mut buf = Vec::new();
2600 let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
2601
2602 self.inner.put_cf(cf, key.as_ref(), value_bytes).map_err(|e| {
2603 ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
2604 info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
2605 operation: DatabaseWriteOperation::PutUpsert,
2606 table_name: T::NAME,
2607 key: key.as_ref().to_vec(),
2608 })))
2609 })
2610 }
2611
2612 pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
2614 let cf = self.provider.get_cf_handle::<T>()?;
2615 self.inner.delete_cf(cf, key.encode().as_ref()).map_err(|e| {
2616 ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
2617 message: e.to_string().into(),
2618 code: -1,
2619 }))
2620 })
2621 }
2622
2623 pub fn iter<T: Table>(&self) -> ProviderResult<RocksTxIter<'_, T>> {
2627 let cf = self.provider.get_cf_handle::<T>()?;
2628 let iter = self.inner.iterator_cf(cf, IteratorMode::Start);
2629 Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2630 }
2631
2632 pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksTxIter<'_, T>> {
2634 let cf = self.provider.get_cf_handle::<T>()?;
2635 let encoded_key = key.encode();
2636 let iter = self
2637 .inner
2638 .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
2639 Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2640 }
2641
2642 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2644 pub fn commit(self) -> ProviderResult<()> {
2645 self.inner.commit().map_err(|e| {
2646 ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2647 message: e.to_string().into(),
2648 code: -1,
2649 }))
2650 })
2651 }
2652
2653 #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2655 pub fn rollback(self) -> ProviderResult<()> {
2656 self.inner.rollback().map_err(|e| {
2657 ProviderError::Database(DatabaseError::Other(format!("rollback failed: {e}")))
2658 })
2659 }
2660}
2661
2662enum RocksDBIterEnum<'db> {
2664 ReadWrite(rocksdb::DBIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2666 ReadOnly(rocksdb::DBIteratorWithThreadMode<'db, DB>),
2668}
2669
2670impl Iterator for RocksDBIterEnum<'_> {
2671 type Item = Result<(Box<[u8]>, Box<[u8]>), rocksdb::Error>;
2672
2673 fn next(&mut self) -> Option<Self::Item> {
2674 match self {
2675 Self::ReadWrite(iter) => iter.next(),
2676 Self::ReadOnly(iter) => iter.next(),
2677 }
2678 }
2679}
2680
2681enum RocksDBRawIterEnum<'db> {
2686 ReadWrite(DBRawIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2688 ReadOnly(DBRawIteratorWithThreadMode<'db, DB>),
2690}
2691
2692impl RocksDBRawIterEnum<'_> {
2693 fn seek(&mut self, key: impl AsRef<[u8]>) {
2695 match self {
2696 Self::ReadWrite(iter) => iter.seek(key),
2697 Self::ReadOnly(iter) => iter.seek(key),
2698 }
2699 }
2700
2701 fn valid(&self) -> bool {
2703 match self {
2704 Self::ReadWrite(iter) => iter.valid(),
2705 Self::ReadOnly(iter) => iter.valid(),
2706 }
2707 }
2708
2709 fn key(&self) -> Option<&[u8]> {
2711 match self {
2712 Self::ReadWrite(iter) => iter.key(),
2713 Self::ReadOnly(iter) => iter.key(),
2714 }
2715 }
2716
2717 fn value(&self) -> Option<&[u8]> {
2719 match self {
2720 Self::ReadWrite(iter) => iter.value(),
2721 Self::ReadOnly(iter) => iter.value(),
2722 }
2723 }
2724
2725 fn next(&mut self) {
2727 match self {
2728 Self::ReadWrite(iter) => iter.next(),
2729 Self::ReadOnly(iter) => iter.next(),
2730 }
2731 }
2732
2733 fn prev(&mut self) {
2735 match self {
2736 Self::ReadWrite(iter) => iter.prev(),
2737 Self::ReadOnly(iter) => iter.prev(),
2738 }
2739 }
2740
2741 fn status(&self) -> Result<(), rocksdb::Error> {
2743 match self {
2744 Self::ReadWrite(iter) => iter.status(),
2745 Self::ReadOnly(iter) => iter.status(),
2746 }
2747 }
2748}
2749
2750pub struct RocksDBIter<'db, T: Table> {
2754 inner: RocksDBIterEnum<'db>,
2755 _marker: std::marker::PhantomData<T>,
2756}
2757
2758impl<T: Table> fmt::Debug for RocksDBIter<'_, T> {
2759 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2760 f.debug_struct("RocksDBIter").field("table", &T::NAME).finish_non_exhaustive()
2761 }
2762}
2763
2764impl<T: Table> Iterator for RocksDBIter<'_, T> {
2765 type Item = ProviderResult<(T::Key, T::Value)>;
2766
2767 fn next(&mut self) -> Option<Self::Item> {
2768 Some(decode_iter_item::<T>(self.inner.next()?))
2769 }
2770}
2771
2772pub struct RocksDBRawIter<'db> {
2776 inner: RocksDBIterEnum<'db>,
2777}
2778
2779impl fmt::Debug for RocksDBRawIter<'_> {
2780 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2781 f.debug_struct("RocksDBRawIter").finish_non_exhaustive()
2782 }
2783}
2784
2785impl Iterator for RocksDBRawIter<'_> {
2786 type Item = ProviderResult<(Box<[u8]>, Box<[u8]>)>;
2787
2788 fn next(&mut self) -> Option<Self::Item> {
2789 match self.inner.next()? {
2790 Ok(kv) => Some(Ok(kv)),
2791 Err(e) => Some(Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2792 message: e.to_string().into(),
2793 code: -1,
2794 })))),
2795 }
2796 }
2797}
2798
2799pub(crate) struct RocksDBRawKeyIter<'db> {
2801 inner: RocksDBRawIterEnum<'db>,
2802}
2803
2804impl fmt::Debug for RocksDBRawKeyIter<'_> {
2805 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2806 f.debug_struct("RocksDBRawKeyIter").finish_non_exhaustive()
2807 }
2808}
2809
2810impl Iterator for RocksDBRawKeyIter<'_> {
2811 type Item = ProviderResult<Box<[u8]>>;
2812
2813 fn next(&mut self) -> Option<Self::Item> {
2814 if !self.inner.valid() {
2815 return self.inner.status().err().map(|e| {
2816 Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2817 message: e.to_string().into(),
2818 code: -1,
2819 })))
2820 })
2821 }
2822
2823 let Some(key) = self.inner.key() else {
2824 return Some(Err(ProviderError::Database(DatabaseError::Decode)))
2825 };
2826 let key = Box::from(key);
2827 self.inner.next();
2828 Some(Ok(key))
2829 }
2830}
2831
2832pub struct RocksTxIter<'tx, T: Table> {
2836 inner: rocksdb::DBIteratorWithThreadMode<'tx, Transaction<'tx, OptimisticTransactionDB>>,
2837 _marker: std::marker::PhantomData<T>,
2838}
2839
2840impl<T: Table> fmt::Debug for RocksTxIter<'_, T> {
2841 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2842 f.debug_struct("RocksTxIter").field("table", &T::NAME).finish_non_exhaustive()
2843 }
2844}
2845
2846impl<T: Table> Iterator for RocksTxIter<'_, T> {
2847 type Item = ProviderResult<(T::Key, T::Value)>;
2848
2849 fn next(&mut self) -> Option<Self::Item> {
2850 Some(decode_iter_item::<T>(self.inner.next()?))
2851 }
2852}
2853
2854fn decode_iter_item<T: Table>(result: RawKVResult) -> ProviderResult<(T::Key, T::Value)> {
2859 let (key_bytes, value_bytes) = result.map_err(|e| {
2860 ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2861 message: e.to_string().into(),
2862 code: -1,
2863 }))
2864 })?;
2865
2866 let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
2867 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2868
2869 let value = T::Value::decompress(&value_bytes)
2870 .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2871
2872 Ok((key, value))
2873}
2874
2875const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
2877 match level {
2878 LogLevel::Fatal => rocksdb::LogLevel::Fatal,
2879 LogLevel::Error => rocksdb::LogLevel::Error,
2880 LogLevel::Warn => rocksdb::LogLevel::Warn,
2881 LogLevel::Notice | LogLevel::Verbose => rocksdb::LogLevel::Info,
2882 LogLevel::Debug | LogLevel::Trace | LogLevel::Extra => rocksdb::LogLevel::Debug,
2883 }
2884}
2885
2886#[cfg(test)]
2887mod tests {
2888 use super::*;
2889 use crate::providers::HistoryInfo;
2890 use alloy_primitives::{Address, Bytes, TxHash, B256};
2891 use reth_db_api::{
2892 models::{
2893 sharded_key::{ShardedKey, NUM_OF_INDICES_IN_SHARD},
2894 storage_sharded_key::StorageShardedKey,
2895 IntegerList,
2896 },
2897 table::Table,
2898 tables,
2899 };
2900 use tempfile::TempDir;
2901
2902 #[test]
2903 fn test_with_default_tables_registers_required_column_families() {
2904 let temp_dir = TempDir::new().unwrap();
2905
2906 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
2908
2909 let tx_hash = TxHash::from(B256::from([1u8; 32]));
2911 provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
2912 assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
2913
2914 let key = ShardedKey::new(Address::ZERO, 100);
2916 let value = IntegerList::default();
2917 provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
2918 assert!(provider.get::<tables::AccountsHistory>(key).unwrap().is_some());
2919
2920 let key = StorageShardedKey::new(Address::ZERO, B256::ZERO, 100);
2922 provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
2923 assert!(provider.get::<tables::StoragesHistory>(key).unwrap().is_some());
2924
2925 let bal_key =
2926 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
2927 let bal_value =
2928 reth_db_api::models::StoredBlockAccessList::new(Bytes::from_static(&[0xc0]));
2929 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
2930 assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
2931 provider
2932 .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
2933 .unwrap();
2934 assert_eq!(
2935 provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
2936 Some(bal_key.number())
2937 );
2938 }
2939
2940 #[test]
2941 fn block_access_lists_store_large_payloads_in_blob_files() {
2942 let temp_dir = TempDir::new().unwrap();
2943 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
2944 let bal_key =
2945 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
2946 let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
2947 0;
2948 DEFAULT_BAL_MIN_BLOB_SIZE as usize +
2949 1
2950 ]));
2951
2952 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
2953 provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
2954
2955 let has_blob_file = std::fs::read_dir(temp_dir.path()).unwrap().any(|entry| {
2956 entry.unwrap().path().extension().is_some_and(|extension| extension == "blob")
2957 });
2958 assert!(has_blob_file);
2959 }
2960
2961 #[derive(Debug)]
2962 struct TestTable;
2963
2964 impl Table for TestTable {
2965 const NAME: &'static str = "TestTable";
2966 const DUPSORT: bool = false;
2967 type Key = u64;
2968 type Value = Vec<u8>;
2969 }
2970
2971 #[test]
2972 fn test_reopens_with_unknown_column_family() {
2973 let temp_dir = TempDir::new().unwrap();
2974 let value = b"test_value".to_vec();
2975
2976 let provider = RocksDBBuilder::new(temp_dir.path())
2977 .with_default_tables()
2978 .with_table::<TestTable>()
2979 .build()
2980 .unwrap();
2981 provider.put::<TestTable>(42, &value).unwrap();
2982 drop(provider);
2983
2984 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
2985 assert_eq!(provider.get::<TestTable>(42).unwrap(), Some(value));
2986 }
2987
2988 #[test]
2989 fn test_reopens_blob_column_family_with_legacy_table_set() {
2990 let temp_dir = TempDir::new().unwrap();
2991 let bal_key =
2992 reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
2993 let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
2994 0;
2995 DEFAULT_BAL_MIN_BLOB_SIZE as usize +
2996 1
2997 ]));
2998
2999 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3000 provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3001 provider
3002 .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
3003 .unwrap();
3004 provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3005 drop(provider);
3006
3007 let provider = RocksDBBuilder::new(temp_dir.path())
3008 .with_table::<tables::TransactionHashNumbers>()
3009 .with_table::<tables::AccountsHistory>()
3010 .with_table::<tables::StoragesHistory>()
3011 .build()
3012 .unwrap();
3013 assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
3014 assert_eq!(
3015 provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
3016 Some(bal_key.number())
3017 );
3018 }
3019
3020 #[test]
3021 fn test_basic_operations() {
3022 let temp_dir = TempDir::new().unwrap();
3023
3024 let provider = RocksDBBuilder::new(temp_dir.path())
3025 .with_table::<TestTable>() .build()
3027 .unwrap();
3028
3029 let key = 42u64;
3030 let value = b"test_value".to_vec();
3031
3032 provider.put::<TestTable>(key, &value).unwrap();
3034
3035 let result = provider.get::<TestTable>(key).unwrap();
3037 assert_eq!(result, Some(value));
3038
3039 provider.delete::<TestTable>(key).unwrap();
3041
3042 assert_eq!(provider.get::<TestTable>(key).unwrap(), None);
3044 }
3045
3046 #[test]
3047 fn test_batch_operations() {
3048 let temp_dir = TempDir::new().unwrap();
3049 let provider =
3050 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3051
3052 provider
3054 .write_batch(|batch| {
3055 for i in 0..10u64 {
3056 let value = format!("value_{i}").into_bytes();
3057 batch.put::<TestTable>(i, &value)?;
3058 }
3059 Ok(())
3060 })
3061 .unwrap();
3062
3063 for i in 0..10u64 {
3065 let value = format!("value_{i}").into_bytes();
3066 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3067 }
3068
3069 provider
3071 .write_batch(|batch| {
3072 for i in 0..10u64 {
3073 batch.delete::<TestTable>(i)?;
3074 }
3075 Ok(())
3076 })
3077 .unwrap();
3078
3079 for i in 0..10u64 {
3081 assert_eq!(provider.get::<TestTable>(i).unwrap(), None);
3082 }
3083 }
3084
3085 #[test]
3086 fn test_with_real_table() {
3087 let temp_dir = TempDir::new().unwrap();
3088 let provider = RocksDBBuilder::new(temp_dir.path())
3089 .with_table::<tables::TransactionHashNumbers>()
3090 .with_metrics()
3091 .build()
3092 .unwrap();
3093
3094 let tx_hash = TxHash::from(B256::from([1u8; 32]));
3095
3096 provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3098 assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3099
3100 provider
3102 .write_batch(|batch| {
3103 for i in 0..10u64 {
3104 let hash = TxHash::from(B256::from([i as u8; 32]));
3105 let value = i * 100;
3106 batch.put::<tables::TransactionHashNumbers>(hash, &value)?;
3107 }
3108 Ok(())
3109 })
3110 .unwrap();
3111
3112 for i in 0..10u64 {
3114 let hash = TxHash::from(B256::from([i as u8; 32]));
3115 assert_eq!(
3116 provider.get::<tables::TransactionHashNumbers>(hash).unwrap(),
3117 Some(i * 100)
3118 );
3119 }
3120 }
3121 #[test]
3122 fn test_statistics_enabled() {
3123 let temp_dir = TempDir::new().unwrap();
3124 let provider = RocksDBBuilder::new(temp_dir.path())
3126 .with_table::<TestTable>()
3127 .with_statistics()
3128 .build()
3129 .unwrap();
3130
3131 for i in 0..10 {
3133 let value = vec![i as u8];
3134 provider.put::<TestTable>(i, &value).unwrap();
3135 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3137 }
3138 }
3139
3140 #[test]
3141 fn test_data_persistence() {
3142 let temp_dir = TempDir::new().unwrap();
3143 let provider =
3144 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3145
3146 let value = vec![42u8; 1000];
3148 for i in 0..100 {
3149 provider.put::<TestTable>(i, &value).unwrap();
3150 }
3151
3152 for i in 0..100 {
3154 assert!(provider.get::<TestTable>(i).unwrap().is_some(), "Data should be readable");
3155 }
3156 }
3157
3158 #[test]
3159 fn test_transaction_read_your_writes() {
3160 let temp_dir = TempDir::new().unwrap();
3161 let provider =
3162 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3163
3164 let tx = provider.tx();
3166
3167 let key = 42u64;
3169 let value = b"test_value".to_vec();
3170 tx.put::<TestTable>(key, &value).unwrap();
3171
3172 let result = tx.get::<TestTable>(key).unwrap();
3174 assert_eq!(
3175 result,
3176 Some(value.clone()),
3177 "Transaction should see its own uncommitted writes"
3178 );
3179
3180 let provider_result = provider.get::<TestTable>(key).unwrap();
3182 assert_eq!(provider_result, None, "Uncommitted data should not be visible outside tx");
3183
3184 tx.commit().unwrap();
3186
3187 let committed_result = provider.get::<TestTable>(key).unwrap();
3189 assert_eq!(committed_result, Some(value), "Committed data should be visible");
3190 }
3191
3192 #[test]
3193 fn test_transaction_rollback() {
3194 let temp_dir = TempDir::new().unwrap();
3195 let provider =
3196 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3197
3198 let key = 100u64;
3200 let initial_value = b"initial".to_vec();
3201 provider.put::<TestTable>(key, &initial_value).unwrap();
3202
3203 let tx = provider.tx();
3205 let new_value = b"modified".to_vec();
3206 tx.put::<TestTable>(key, &new_value).unwrap();
3207
3208 assert_eq!(tx.get::<TestTable>(key).unwrap(), Some(new_value));
3210
3211 tx.rollback().unwrap();
3213
3214 let result = provider.get::<TestTable>(key).unwrap();
3216 assert_eq!(result, Some(initial_value), "Rollback should preserve original data");
3217 }
3218
3219 #[test]
3220 fn test_transaction_iterator() {
3221 let temp_dir = TempDir::new().unwrap();
3222 let provider =
3223 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3224
3225 let tx = provider.tx();
3227
3228 for i in 0..5u64 {
3230 let value = format!("value_{i}").into_bytes();
3231 tx.put::<TestTable>(i, &value).unwrap();
3232 }
3233
3234 let mut count = 0;
3236 for result in tx.iter::<TestTable>().unwrap() {
3237 let (key, value) = result.unwrap();
3238 assert_eq!(value, format!("value_{key}").into_bytes());
3239 count += 1;
3240 }
3241 assert_eq!(count, 5, "Iterator should see all uncommitted writes");
3242
3243 tx.commit().unwrap();
3245 }
3246
3247 #[test]
3248 fn test_batch_manual_commit() {
3249 let temp_dir = TempDir::new().unwrap();
3250 let provider =
3251 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3252
3253 let mut batch = provider.batch();
3255
3256 for i in 0..10u64 {
3258 let value = format!("batch_value_{i}").into_bytes();
3259 batch.put::<TestTable>(i, &value).unwrap();
3260 }
3261
3262 assert_eq!(batch.len(), 10);
3264 assert!(!batch.is_empty());
3265
3266 assert_eq!(provider.get::<TestTable>(0).unwrap(), None);
3268
3269 batch.commit().unwrap();
3271
3272 for i in 0..10u64 {
3274 let value = format!("batch_value_{i}").into_bytes();
3275 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3276 }
3277 }
3278
3279 #[test]
3280 fn test_first_and_last_entry() {
3281 let temp_dir = TempDir::new().unwrap();
3282 let provider =
3283 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3284
3285 assert_eq!(provider.first::<TestTable>().unwrap(), None);
3287 assert_eq!(provider.last::<TestTable>().unwrap(), None);
3288
3289 provider.put::<TestTable>(10, &b"value_10".to_vec()).unwrap();
3291 provider.put::<TestTable>(20, &b"value_20".to_vec()).unwrap();
3292 provider.put::<TestTable>(5, &b"value_5".to_vec()).unwrap();
3293
3294 let first = provider.first::<TestTable>().unwrap();
3296 assert_eq!(first, Some((5, b"value_5".to_vec())));
3297
3298 let last = provider.last::<TestTable>().unwrap();
3300 assert_eq!(last, Some((20, b"value_20".to_vec())));
3301 }
3302
3303 #[test]
3307 fn test_account_history_info_pruned_before_first_entry() {
3308 let temp_dir = TempDir::new().unwrap();
3309 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3310
3311 let address = Address::from([0x42; 20]);
3312
3313 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3315 let shard_key = ShardedKey::new(address, u64::MAX);
3316 provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3317
3318 let result =
3323 provider.snapshot().account_history_info(address, 50, Some(100), u64::MAX).unwrap();
3324 assert_eq!(result, HistoryInfo::InChangeset(100));
3325 }
3326
3327 #[test]
3329 fn test_account_history_info_read_only_and_catch_up() {
3330 let temp_dir = TempDir::new().unwrap();
3331 let address = Address::from([0x42; 20]);
3332 let chunk = IntegerList::new([100, 200, 300]).unwrap();
3333 let shard_key = ShardedKey::new(address, u64::MAX);
3334
3335 let rw_provider =
3337 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3338 rw_provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3339
3340 let ro_provider = RocksDBBuilder::new(temp_dir.path())
3342 .with_default_tables()
3343 .with_read_only(true)
3344 .build()
3345 .unwrap();
3346
3347 let result =
3348 ro_provider.snapshot().account_history_info(address, 200, None, u64::MAX).unwrap();
3349 assert_eq!(result, HistoryInfo::InChangeset(200));
3350
3351 let result =
3352 ro_provider.snapshot().account_history_info(address, 50, None, u64::MAX).unwrap();
3353 assert_eq!(result, HistoryInfo::NotYetWritten);
3354
3355 let result =
3356 ro_provider.snapshot().account_history_info(address, 400, None, u64::MAX).unwrap();
3357 assert_eq!(result, HistoryInfo::InPlainState);
3358
3359 let address2 = Address::from([0x43; 20]);
3361 let chunk2 = IntegerList::new([500, 600]).unwrap();
3362 let shard_key2 = ShardedKey::new(address2, u64::MAX);
3363 rw_provider.put::<tables::AccountsHistory>(shard_key2, &chunk2).unwrap();
3364
3365 let result =
3367 ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3368 assert_eq!(result, HistoryInfo::NotYetWritten);
3369
3370 ro_provider.try_catch_up_with_primary().unwrap();
3372
3373 let result =
3374 ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3375 assert_eq!(result, HistoryInfo::InChangeset(500));
3376 }
3377
3378 #[test]
3379 fn test_account_history_info_ignores_blocks_above_visible_tip() {
3380 let temp_dir = TempDir::new().unwrap();
3381 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3382
3383 let address = Address::from([0x42; 20]);
3384
3385 provider
3386 .put::<tables::AccountsHistory>(
3387 ShardedKey::new(address, 110),
3388 &IntegerList::new([100, 110]).unwrap(),
3389 )
3390 .unwrap();
3391 provider
3392 .put::<tables::AccountsHistory>(
3393 ShardedKey::new(address, u64::MAX),
3394 &IntegerList::new([200, 210]).unwrap(),
3395 )
3396 .unwrap();
3397
3398 let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3399 assert_eq!(result, HistoryInfo::InPlainState);
3400 }
3401
3402 #[test]
3403 fn test_account_history_info_mixed_shard_respects_visible_tip() {
3404 let temp_dir = TempDir::new().unwrap();
3405 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3406
3407 let address = Address::from([0x42; 20]);
3408 provider
3409 .put::<tables::AccountsHistory>(
3410 ShardedKey::new(address, u64::MAX),
3411 &IntegerList::new([100, 150, 300]).unwrap(),
3412 )
3413 .unwrap();
3414
3415 let result = provider.snapshot().account_history_info(address, 120, None, 200).unwrap();
3416 assert_eq!(result, HistoryInfo::InChangeset(150));
3417
3418 let result = provider.snapshot().account_history_info(address, 201, None, 200).unwrap();
3419 assert_eq!(result, HistoryInfo::InPlainState);
3420 }
3421
3422 #[test]
3423 fn test_account_history_info_only_stale_entries_use_fallback() {
3424 let temp_dir = TempDir::new().unwrap();
3425 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3426
3427 let address = Address::from([0x42; 20]);
3428 provider
3429 .put::<tables::AccountsHistory>(
3430 ShardedKey::new(address, u64::MAX),
3431 &IntegerList::new([200, 210]).unwrap(),
3432 )
3433 .unwrap();
3434
3435 let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3436 assert_eq!(result, HistoryInfo::NotYetWritten);
3437
3438 let result =
3439 provider.snapshot().account_history_info(address, 150, Some(100), 150).unwrap();
3440 assert_eq!(result, HistoryInfo::MaybeInPlainState);
3441 }
3442
3443 #[test]
3444 fn test_account_history_shard_split_at_boundary() {
3445 let temp_dir = TempDir::new().unwrap();
3446 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3447
3448 let address = Address::from([0x42; 20]);
3449 let limit = NUM_OF_INDICES_IN_SHARD;
3450
3451 let indices: Vec<u64> = (0..=(limit as u64)).collect();
3453 let mut batch = provider.batch();
3454 batch.append_account_history_shard(address, indices).unwrap();
3455 batch.commit().unwrap();
3456
3457 let completed_key = ShardedKey::new(address, (limit - 1) as u64);
3459 let sentinel_key = ShardedKey::new(address, u64::MAX);
3460
3461 let completed_shard = provider.get::<tables::AccountsHistory>(completed_key).unwrap();
3462 let sentinel_shard = provider.get::<tables::AccountsHistory>(sentinel_key).unwrap();
3463
3464 assert!(completed_shard.is_some(), "completed shard should exist");
3465 assert!(sentinel_shard.is_some(), "sentinel shard should exist");
3466
3467 let completed_shard = completed_shard.unwrap();
3468 let sentinel_shard = sentinel_shard.unwrap();
3469
3470 assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
3471 assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
3472 }
3473
3474 #[test]
3475 fn test_account_history_multiple_shard_splits() {
3476 let temp_dir = TempDir::new().unwrap();
3477 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3478
3479 let address = Address::from([0x43; 20]);
3480 let limit = NUM_OF_INDICES_IN_SHARD;
3481
3482 let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
3484 let mut batch = provider.batch();
3485 batch.append_account_history_shard(address, first_batch_indices).unwrap();
3486 batch.commit().unwrap();
3487
3488 let sentinel_key = ShardedKey::new(address, u64::MAX);
3490 let shard = provider.get::<tables::AccountsHistory>(sentinel_key.clone()).unwrap();
3491 assert!(shard.is_some());
3492 assert_eq!(shard.unwrap().len(), limit as u64);
3493
3494 let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
3496 let mut batch = provider.batch();
3497 batch.append_account_history_shard(address, second_batch_indices).unwrap();
3498 batch.commit().unwrap();
3499
3500 let first_completed = ShardedKey::new(address, (limit - 1) as u64);
3502 let second_completed = ShardedKey::new(address, (2 * limit - 1) as u64);
3503
3504 assert!(
3505 provider.get::<tables::AccountsHistory>(first_completed).unwrap().is_some(),
3506 "first completed shard should exist"
3507 );
3508 assert!(
3509 provider.get::<tables::AccountsHistory>(second_completed).unwrap().is_some(),
3510 "second completed shard should exist"
3511 );
3512 assert!(
3513 provider.get::<tables::AccountsHistory>(sentinel_key).unwrap().is_some(),
3514 "sentinel shard should exist"
3515 );
3516 }
3517
3518 #[test]
3519 fn test_storage_history_shard_split_at_boundary() {
3520 let temp_dir = TempDir::new().unwrap();
3521 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3522
3523 let address = Address::from([0x44; 20]);
3524 let slot = B256::from([0x55; 32]);
3525 let limit = NUM_OF_INDICES_IN_SHARD;
3526
3527 let indices: Vec<u64> = (0..=(limit as u64)).collect();
3529 let mut batch = provider.batch();
3530 batch.append_storage_history_shard(address, slot, indices).unwrap();
3531 batch.commit().unwrap();
3532
3533 let completed_key = StorageShardedKey::new(address, slot, (limit - 1) as u64);
3535 let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
3536
3537 let completed_shard = provider.get::<tables::StoragesHistory>(completed_key).unwrap();
3538 let sentinel_shard = provider.get::<tables::StoragesHistory>(sentinel_key).unwrap();
3539
3540 assert!(completed_shard.is_some(), "completed shard should exist");
3541 assert!(sentinel_shard.is_some(), "sentinel shard should exist");
3542
3543 let completed_shard = completed_shard.unwrap();
3544 let sentinel_shard = sentinel_shard.unwrap();
3545
3546 assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
3547 assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
3548 }
3549
3550 #[test]
3551 fn test_storage_history_multiple_shard_splits() {
3552 let temp_dir = TempDir::new().unwrap();
3553 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3554
3555 let address = Address::from([0x46; 20]);
3556 let slot = B256::from([0x57; 32]);
3557 let limit = NUM_OF_INDICES_IN_SHARD;
3558
3559 let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
3561 let mut batch = provider.batch();
3562 batch.append_storage_history_shard(address, slot, first_batch_indices).unwrap();
3563 batch.commit().unwrap();
3564
3565 let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
3567 let shard = provider.get::<tables::StoragesHistory>(sentinel_key.clone()).unwrap();
3568 assert!(shard.is_some());
3569 assert_eq!(shard.unwrap().len(), limit as u64);
3570
3571 let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
3573 let mut batch = provider.batch();
3574 batch.append_storage_history_shard(address, slot, second_batch_indices).unwrap();
3575 batch.commit().unwrap();
3576
3577 let first_completed = StorageShardedKey::new(address, slot, (limit - 1) as u64);
3579 let second_completed = StorageShardedKey::new(address, slot, (2 * limit - 1) as u64);
3580
3581 assert!(
3582 provider.get::<tables::StoragesHistory>(first_completed).unwrap().is_some(),
3583 "first completed shard should exist"
3584 );
3585 assert!(
3586 provider.get::<tables::StoragesHistory>(second_completed).unwrap().is_some(),
3587 "second completed shard should exist"
3588 );
3589 assert!(
3590 provider.get::<tables::StoragesHistory>(sentinel_key).unwrap().is_some(),
3591 "sentinel shard should exist"
3592 );
3593 }
3594
3595 #[test]
3596 fn test_clear_table() {
3597 let temp_dir = TempDir::new().unwrap();
3598 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3599
3600 let address = Address::from([0x42; 20]);
3601 let key = ShardedKey::new(address, u64::MAX);
3602 let blocks = BlockNumberList::new_pre_sorted([1, 2, 3]);
3603
3604 provider.put::<tables::AccountsHistory>(key.clone(), &blocks).unwrap();
3605 assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
3606
3607 provider.clear::<tables::AccountsHistory>().unwrap();
3608
3609 assert!(
3610 provider.get::<tables::AccountsHistory>(key).unwrap().is_none(),
3611 "table should be empty after clear"
3612 );
3613 assert!(
3614 provider.first::<tables::AccountsHistory>().unwrap().is_none(),
3615 "first() should return None after clear"
3616 );
3617 }
3618
3619 #[test]
3620 fn test_clear_empty_table() {
3621 let temp_dir = TempDir::new().unwrap();
3622 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3623
3624 assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
3625
3626 provider.clear::<tables::AccountsHistory>().unwrap();
3627
3628 assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
3629 }
3630
3631 #[test]
3632 fn test_unwind_account_history_to_basic() {
3633 let temp_dir = TempDir::new().unwrap();
3634 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3635
3636 let address = Address::from([0x42; 20]);
3637
3638 let mut batch = provider.batch();
3640 batch.append_account_history_shard(address, 0..=10).unwrap();
3641 batch.commit().unwrap();
3642
3643 let key = ShardedKey::new(address, u64::MAX);
3645 let result = provider.get::<tables::AccountsHistory>(key.clone()).unwrap();
3646 assert!(result.is_some());
3647 let blocks: Vec<u64> = result.unwrap().iter().collect();
3648 assert_eq!(blocks, (0..=10).collect::<Vec<_>>());
3649
3650 let mut batch = provider.batch();
3652 batch.unwind_account_history_to(address, 5).unwrap();
3653 batch.commit().unwrap();
3654
3655 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3657 assert!(result.is_some());
3658 let blocks: Vec<u64> = result.unwrap().iter().collect();
3659 assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
3660 }
3661
3662 #[test]
3663 fn test_unwind_account_history_to_removes_all() {
3664 let temp_dir = TempDir::new().unwrap();
3665 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3666
3667 let address = Address::from([0x42; 20]);
3668
3669 let mut batch = provider.batch();
3671 batch.append_account_history_shard(address, 5..=10).unwrap();
3672 batch.commit().unwrap();
3673
3674 let mut batch = provider.batch();
3676 batch.unwind_account_history_to(address, 4).unwrap();
3677 batch.commit().unwrap();
3678
3679 let key = ShardedKey::new(address, u64::MAX);
3681 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3682 assert!(result.is_none(), "Should have no data after full unwind");
3683 }
3684
3685 #[test]
3686 fn test_unwind_account_history_to_no_op() {
3687 let temp_dir = TempDir::new().unwrap();
3688 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3689
3690 let address = Address::from([0x42; 20]);
3691
3692 let mut batch = provider.batch();
3694 batch.append_account_history_shard(address, 0..=5).unwrap();
3695 batch.commit().unwrap();
3696
3697 let mut batch = provider.batch();
3699 batch.unwind_account_history_to(address, 10).unwrap();
3700 batch.commit().unwrap();
3701
3702 let key = ShardedKey::new(address, u64::MAX);
3704 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3705 assert!(result.is_some());
3706 let blocks: Vec<u64> = result.unwrap().iter().collect();
3707 assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
3708 }
3709
3710 #[test]
3711 fn test_unwind_account_history_to_block_zero() {
3712 let temp_dir = TempDir::new().unwrap();
3713 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3714
3715 let address = Address::from([0x42; 20]);
3716
3717 let mut batch = provider.batch();
3719 batch.append_account_history_shard(address, 0..=5).unwrap();
3720 batch.commit().unwrap();
3721
3722 let mut batch = provider.batch();
3725 batch.unwind_account_history_to(address, 0).unwrap();
3726 batch.commit().unwrap();
3727
3728 let key = ShardedKey::new(address, u64::MAX);
3730 let result = provider.get::<tables::AccountsHistory>(key).unwrap();
3731 assert!(result.is_some());
3732 let blocks: Vec<u64> = result.unwrap().iter().collect();
3733 assert_eq!(blocks, vec![0]);
3734 }
3735
3736 #[test]
3737 fn test_unwind_account_history_to_multi_shard() {
3738 let temp_dir = TempDir::new().unwrap();
3739 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3740
3741 let address = Address::from([0x42; 20]);
3742
3743 let mut batch = provider.batch();
3746
3747 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
3749 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
3750
3751 let shard2 = BlockNumberList::new_pre_sorted(51..=100);
3753 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
3754
3755 batch.commit().unwrap();
3756
3757 let shards = provider.account_history_shards(address).unwrap();
3759 assert_eq!(shards.len(), 2);
3760
3761 let mut batch = provider.batch();
3763 batch.unwind_account_history_to(address, 75).unwrap();
3764 batch.commit().unwrap();
3765
3766 let shards = provider.account_history_shards(address).unwrap();
3768 assert_eq!(shards.len(), 2);
3769
3770 assert_eq!(shards[0].0.highest_block_number, 50);
3772 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
3773
3774 assert_eq!(shards[1].0.highest_block_number, u64::MAX);
3776 assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
3777 }
3778
3779 #[test]
3780 fn test_unwind_account_history_to_multi_shard_boundary_empty() {
3781 let temp_dir = TempDir::new().unwrap();
3782 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3783
3784 let address = Address::from([0x42; 20]);
3785
3786 let mut batch = provider.batch();
3788
3789 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
3791 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
3792
3793 let shard2 = BlockNumberList::new_pre_sorted(75..=100);
3795 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
3796
3797 batch.commit().unwrap();
3798
3799 let mut batch = provider.batch();
3801 batch.unwind_account_history_to(address, 60).unwrap();
3802 batch.commit().unwrap();
3803
3804 let shards = provider.account_history_shards(address).unwrap();
3806 assert_eq!(shards.len(), 1);
3807 assert_eq!(shards[0].0.highest_block_number, u64::MAX);
3808 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
3809 }
3810
3811 #[test]
3812 fn test_account_history_shards_iterator() {
3813 let temp_dir = TempDir::new().unwrap();
3814 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3815
3816 let address = Address::from([0x42; 20]);
3817 let other_address = Address::from([0x43; 20]);
3818
3819 let mut batch = provider.batch();
3821 batch.append_account_history_shard(address, 0..=5).unwrap();
3822 batch.append_account_history_shard(other_address, 10..=15).unwrap();
3823 batch.commit().unwrap();
3824
3825 let shards = provider.account_history_shards(address).unwrap();
3827 assert_eq!(shards.len(), 1);
3828 assert_eq!(shards[0].0.key, address);
3829
3830 let shards = provider.account_history_shards(other_address).unwrap();
3832 assert_eq!(shards.len(), 1);
3833 assert_eq!(shards[0].0.key, other_address);
3834
3835 let non_existent = Address::from([0x99; 20]);
3837 let shards = provider.account_history_shards(non_existent).unwrap();
3838 assert!(shards.is_empty());
3839 }
3840
3841 #[test]
3842 fn test_clear_account_history() {
3843 let temp_dir = TempDir::new().unwrap();
3844 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3845
3846 let address = Address::from([0x42; 20]);
3847
3848 let mut batch = provider.batch();
3850 batch.append_account_history_shard(address, 0..=10).unwrap();
3851 batch.commit().unwrap();
3852
3853 let mut batch = provider.batch();
3855 batch.clear_account_history(address).unwrap();
3856 batch.commit().unwrap();
3857
3858 let shards = provider.account_history_shards(address).unwrap();
3860 assert!(shards.is_empty(), "All shards should be deleted");
3861 }
3862
3863 #[test]
3864 fn test_unwind_non_sentinel_boundary() {
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();
3872
3873 let shard1 = BlockNumberList::new_pre_sorted(1..=50);
3875 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
3876
3877 let shard2 = BlockNumberList::new_pre_sorted(51..=100);
3879 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 100), &shard2).unwrap();
3880
3881 let shard3 = BlockNumberList::new_pre_sorted(101..=150);
3883 batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard3).unwrap();
3884
3885 batch.commit().unwrap();
3886
3887 let shards = provider.account_history_shards(address).unwrap();
3889 assert_eq!(shards.len(), 3);
3890
3891 let mut batch = provider.batch();
3893 batch.unwind_account_history_to(address, 75).unwrap();
3894 batch.commit().unwrap();
3895
3896 let shards = provider.account_history_shards(address).unwrap();
3898 assert_eq!(shards.len(), 2);
3899
3900 assert_eq!(shards[0].0.highest_block_number, 50);
3902 assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
3903
3904 assert_eq!(shards[1].0.highest_block_number, u64::MAX);
3906 assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
3907 }
3908
3909 #[test]
3910 fn test_batch_auto_commit_on_threshold() {
3911 let temp_dir = TempDir::new().unwrap();
3912 let provider =
3913 RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3914
3915 let mut batch = RocksDBBatch {
3917 provider: &provider,
3918 inner: WriteBatchWithTransaction::<true>::default(),
3919 buf: Vec::new(),
3920 auto_commit_threshold: Some(1024), };
3922
3923 for i in 0..100u64 {
3926 let value = format!("value_{i:04}").into_bytes();
3927 batch.put::<TestTable>(i, &value).unwrap();
3928 }
3929
3930 let first_visible = provider.get::<TestTable>(0).unwrap();
3933 assert!(first_visible.is_some(), "Auto-committed data should be visible");
3934
3935 batch.commit().unwrap();
3937
3938 for i in 0..100u64 {
3940 let value = format!("value_{i:04}").into_bytes();
3941 assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3942 }
3943 }
3944
3945 struct AccountPruneCase {
3949 name: &'static str,
3950 initial_shards: &'static [(u64, &'static [u64])],
3951 prune_to: u64,
3952 expected_outcome: PruneShardOutcome,
3953 expected_shards: &'static [(u64, &'static [u64])],
3954 }
3955
3956 struct StoragePruneCase {
3958 name: &'static str,
3959 initial_shards: &'static [(u64, &'static [u64])],
3960 prune_to: u64,
3961 expected_outcome: PruneShardOutcome,
3962 expected_shards: &'static [(u64, &'static [u64])],
3963 }
3964
3965 #[test]
3966 fn test_prune_account_history_cases() {
3967 const MAX: u64 = u64::MAX;
3968 const CASES: &[AccountPruneCase] = &[
3969 AccountPruneCase {
3970 name: "single_shard_truncate",
3971 initial_shards: &[(MAX, &[10, 20, 30, 40])],
3972 prune_to: 25,
3973 expected_outcome: PruneShardOutcome::Updated,
3974 expected_shards: &[(MAX, &[30, 40])],
3975 },
3976 AccountPruneCase {
3977 name: "single_shard_delete_all",
3978 initial_shards: &[(MAX, &[10, 20])],
3979 prune_to: 20,
3980 expected_outcome: PruneShardOutcome::Deleted,
3981 expected_shards: &[],
3982 },
3983 AccountPruneCase {
3984 name: "single_shard_noop",
3985 initial_shards: &[(MAX, &[10, 20])],
3986 prune_to: 5,
3987 expected_outcome: PruneShardOutcome::Unchanged,
3988 expected_shards: &[(MAX, &[10, 20])],
3989 },
3990 AccountPruneCase {
3991 name: "no_shards",
3992 initial_shards: &[],
3993 prune_to: 100,
3994 expected_outcome: PruneShardOutcome::Unchanged,
3995 expected_shards: &[],
3996 },
3997 AccountPruneCase {
3998 name: "multi_shard_truncate_first",
3999 initial_shards: &[(30, &[10, 20, 30]), (MAX, &[40, 50, 60])],
4000 prune_to: 25,
4001 expected_outcome: PruneShardOutcome::Updated,
4002 expected_shards: &[(30, &[30]), (MAX, &[40, 50, 60])],
4003 },
4004 AccountPruneCase {
4005 name: "delete_first_shard_sentinel_unchanged",
4006 initial_shards: &[(20, &[10, 20]), (MAX, &[30, 40])],
4007 prune_to: 20,
4008 expected_outcome: PruneShardOutcome::Deleted,
4009 expected_shards: &[(MAX, &[30, 40])],
4010 },
4011 AccountPruneCase {
4012 name: "multi_shard_delete_all_but_last",
4013 initial_shards: &[(10, &[5, 10]), (20, &[15, 20]), (MAX, &[25, 30])],
4014 prune_to: 22,
4015 expected_outcome: PruneShardOutcome::Deleted,
4016 expected_shards: &[(MAX, &[25, 30])],
4017 },
4018 AccountPruneCase {
4019 name: "mid_shard_preserves_key",
4020 initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4021 prune_to: 25,
4022 expected_outcome: PruneShardOutcome::Updated,
4023 expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4024 },
4025 AccountPruneCase {
4027 name: "equiv_delete_early_shards_keep_sentinel",
4028 initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4029 prune_to: 55,
4030 expected_outcome: PruneShardOutcome::Deleted,
4031 expected_shards: &[(MAX, &[60, 70])],
4032 },
4033 AccountPruneCase {
4034 name: "equiv_sentinel_becomes_empty_with_prev",
4035 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4036 prune_to: 40,
4037 expected_outcome: PruneShardOutcome::Deleted,
4038 expected_shards: &[(MAX, &[50])],
4039 },
4040 AccountPruneCase {
4041 name: "equiv_all_shards_become_empty",
4042 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4043 prune_to: 51,
4044 expected_outcome: PruneShardOutcome::Deleted,
4045 expected_shards: &[],
4046 },
4047 AccountPruneCase {
4048 name: "equiv_non_sentinel_last_shard_promoted",
4049 initial_shards: &[(100, &[50, 75, 100])],
4050 prune_to: 60,
4051 expected_outcome: PruneShardOutcome::Updated,
4052 expected_shards: &[(MAX, &[75, 100])],
4053 },
4054 AccountPruneCase {
4055 name: "equiv_filter_within_shard",
4056 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4057 prune_to: 25,
4058 expected_outcome: PruneShardOutcome::Updated,
4059 expected_shards: &[(MAX, &[30, 40])],
4060 },
4061 AccountPruneCase {
4062 name: "equiv_multi_shard_partial_delete",
4063 initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4064 prune_to: 35,
4065 expected_outcome: PruneShardOutcome::Deleted,
4066 expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4067 },
4068 ];
4069
4070 let address = Address::from([0x42; 20]);
4071
4072 for case in CASES {
4073 let temp_dir = TempDir::new().unwrap();
4074 let provider =
4075 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4076
4077 let mut batch = provider.batch();
4079 for (highest, blocks) in case.initial_shards {
4080 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4081 batch
4082 .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4083 .unwrap();
4084 }
4085 batch.commit().unwrap();
4086
4087 let mut batch = provider.batch();
4089 let outcome = batch.prune_account_history_to(address, case.prune_to).unwrap();
4090 batch.commit().unwrap();
4091
4092 assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4094
4095 let shards = provider.account_history_shards(address).unwrap();
4097 assert_eq!(
4098 shards.len(),
4099 case.expected_shards.len(),
4100 "case '{}': wrong shard count",
4101 case.name
4102 );
4103 for (i, ((key, blocks), (exp_key, exp_blocks))) in
4104 shards.iter().zip(case.expected_shards.iter()).enumerate()
4105 {
4106 assert_eq!(
4107 key.highest_block_number, *exp_key,
4108 "case '{}': shard {} wrong key",
4109 case.name, i
4110 );
4111 assert_eq!(
4112 blocks.iter().collect::<Vec<_>>(),
4113 *exp_blocks,
4114 "case '{}': shard {} wrong blocks",
4115 case.name,
4116 i
4117 );
4118 }
4119 }
4120 }
4121
4122 #[test]
4123 fn test_prune_storage_history_cases() {
4124 const MAX: u64 = u64::MAX;
4125 const CASES: &[StoragePruneCase] = &[
4126 StoragePruneCase {
4127 name: "single_shard_truncate",
4128 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4129 prune_to: 25,
4130 expected_outcome: PruneShardOutcome::Updated,
4131 expected_shards: &[(MAX, &[30, 40])],
4132 },
4133 StoragePruneCase {
4134 name: "single_shard_delete_all",
4135 initial_shards: &[(MAX, &[10, 20])],
4136 prune_to: 20,
4137 expected_outcome: PruneShardOutcome::Deleted,
4138 expected_shards: &[],
4139 },
4140 StoragePruneCase {
4141 name: "noop",
4142 initial_shards: &[(MAX, &[10, 20])],
4143 prune_to: 5,
4144 expected_outcome: PruneShardOutcome::Unchanged,
4145 expected_shards: &[(MAX, &[10, 20])],
4146 },
4147 StoragePruneCase {
4148 name: "no_shards",
4149 initial_shards: &[],
4150 prune_to: 100,
4151 expected_outcome: PruneShardOutcome::Unchanged,
4152 expected_shards: &[],
4153 },
4154 StoragePruneCase {
4155 name: "mid_shard_preserves_key",
4156 initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4157 prune_to: 25,
4158 expected_outcome: PruneShardOutcome::Updated,
4159 expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4160 },
4161 StoragePruneCase {
4163 name: "equiv_sentinel_promotion",
4164 initial_shards: &[(100, &[50, 75, 100])],
4165 prune_to: 60,
4166 expected_outcome: PruneShardOutcome::Updated,
4167 expected_shards: &[(MAX, &[75, 100])],
4168 },
4169 StoragePruneCase {
4170 name: "equiv_delete_early_shards_keep_sentinel",
4171 initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4172 prune_to: 55,
4173 expected_outcome: PruneShardOutcome::Deleted,
4174 expected_shards: &[(MAX, &[60, 70])],
4175 },
4176 StoragePruneCase {
4177 name: "equiv_sentinel_becomes_empty_with_prev",
4178 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4179 prune_to: 40,
4180 expected_outcome: PruneShardOutcome::Deleted,
4181 expected_shards: &[(MAX, &[50])],
4182 },
4183 StoragePruneCase {
4184 name: "equiv_all_shards_become_empty",
4185 initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4186 prune_to: 51,
4187 expected_outcome: PruneShardOutcome::Deleted,
4188 expected_shards: &[],
4189 },
4190 StoragePruneCase {
4191 name: "equiv_filter_within_shard",
4192 initial_shards: &[(MAX, &[10, 20, 30, 40])],
4193 prune_to: 25,
4194 expected_outcome: PruneShardOutcome::Updated,
4195 expected_shards: &[(MAX, &[30, 40])],
4196 },
4197 StoragePruneCase {
4198 name: "equiv_multi_shard_partial_delete",
4199 initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4200 prune_to: 35,
4201 expected_outcome: PruneShardOutcome::Deleted,
4202 expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4203 },
4204 ];
4205
4206 let address = Address::from([0x42; 20]);
4207 let storage_key = B256::from([0x01; 32]);
4208
4209 for case in CASES {
4210 let temp_dir = TempDir::new().unwrap();
4211 let provider =
4212 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4213
4214 let mut batch = provider.batch();
4216 for (highest, blocks) in case.initial_shards {
4217 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4218 let key = if *highest == MAX {
4219 StorageShardedKey::last(address, storage_key)
4220 } else {
4221 StorageShardedKey::new(address, storage_key, *highest)
4222 };
4223 batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4224 }
4225 batch.commit().unwrap();
4226
4227 let mut batch = provider.batch();
4229 let outcome =
4230 batch.prune_storage_history_to(address, storage_key, case.prune_to).unwrap();
4231 batch.commit().unwrap();
4232
4233 assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4235
4236 let shards = provider.storage_history_shards(address, storage_key).unwrap();
4238 assert_eq!(
4239 shards.len(),
4240 case.expected_shards.len(),
4241 "case '{}': wrong shard count",
4242 case.name
4243 );
4244 for (i, ((key, blocks), (exp_key, exp_blocks))) in
4245 shards.iter().zip(case.expected_shards.iter()).enumerate()
4246 {
4247 assert_eq!(
4248 key.sharded_key.highest_block_number, *exp_key,
4249 "case '{}': shard {} wrong key",
4250 case.name, i
4251 );
4252 assert_eq!(
4253 blocks.iter().collect::<Vec<_>>(),
4254 *exp_blocks,
4255 "case '{}': shard {} wrong blocks",
4256 case.name,
4257 i
4258 );
4259 }
4260 }
4261 }
4262
4263 #[test]
4264 fn test_prune_storage_history_does_not_affect_other_slots() {
4265 let temp_dir = TempDir::new().unwrap();
4266 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4267
4268 let address = Address::from([0x42; 20]);
4269 let slot1 = B256::from([0x01; 32]);
4270 let slot2 = B256::from([0x02; 32]);
4271
4272 let mut batch = provider.batch();
4274 batch
4275 .put::<tables::StoragesHistory>(
4276 StorageShardedKey::last(address, slot1),
4277 &BlockNumberList::new_pre_sorted([10u64, 20]),
4278 )
4279 .unwrap();
4280 batch
4281 .put::<tables::StoragesHistory>(
4282 StorageShardedKey::last(address, slot2),
4283 &BlockNumberList::new_pre_sorted([30u64, 40]),
4284 )
4285 .unwrap();
4286 batch.commit().unwrap();
4287
4288 let mut batch = provider.batch();
4290 let outcome = batch.prune_storage_history_to(address, slot1, 20).unwrap();
4291 batch.commit().unwrap();
4292
4293 assert_eq!(outcome, PruneShardOutcome::Deleted);
4294
4295 let shards1 = provider.storage_history_shards(address, slot1).unwrap();
4297 assert!(shards1.is_empty());
4298
4299 let shards2 = provider.storage_history_shards(address, slot2).unwrap();
4301 assert_eq!(shards2.len(), 1);
4302 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![30, 40]);
4303 }
4304
4305 #[test]
4306 fn test_prune_invariants() {
4307 let address = Address::from([0x42; 20]);
4309 let storage_key = B256::from([0x01; 32]);
4310
4311 #[expect(clippy::type_complexity)]
4313 let invariant_cases: &[(&[(u64, &[u64])], u64)] = &[
4314 (&[(10, &[5, 10]), (20, &[15, 20]), (u64::MAX, &[25, 30])], 20),
4316 (&[(100, &[50, 100])], 60),
4318 ];
4319
4320 for (initial_shards, prune_to) in invariant_cases {
4321 {
4323 let temp_dir = TempDir::new().unwrap();
4324 let provider =
4325 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4326
4327 let mut batch = provider.batch();
4328 for (highest, blocks) in *initial_shards {
4329 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4330 batch
4331 .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4332 .unwrap();
4333 }
4334 batch.commit().unwrap();
4335
4336 let mut batch = provider.batch();
4337 batch.prune_account_history_to(address, *prune_to).unwrap();
4338 batch.commit().unwrap();
4339
4340 let shards = provider.account_history_shards(address).unwrap();
4341
4342 for (key, blocks) in &shards {
4344 assert!(
4345 !blocks.is_empty(),
4346 "Account: empty shard at key {}",
4347 key.highest_block_number
4348 );
4349 }
4350
4351 if !shards.is_empty() {
4353 let last = shards.last().unwrap();
4354 assert_eq!(
4355 last.0.highest_block_number,
4356 u64::MAX,
4357 "Account: last shard must be sentinel"
4358 );
4359 }
4360 }
4361
4362 {
4364 let temp_dir = TempDir::new().unwrap();
4365 let provider =
4366 RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4367
4368 let mut batch = provider.batch();
4369 for (highest, blocks) in *initial_shards {
4370 let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4371 let key = if *highest == u64::MAX {
4372 StorageShardedKey::last(address, storage_key)
4373 } else {
4374 StorageShardedKey::new(address, storage_key, *highest)
4375 };
4376 batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4377 }
4378 batch.commit().unwrap();
4379
4380 let mut batch = provider.batch();
4381 batch.prune_storage_history_to(address, storage_key, *prune_to).unwrap();
4382 batch.commit().unwrap();
4383
4384 let shards = provider.storage_history_shards(address, storage_key).unwrap();
4385
4386 for (key, blocks) in &shards {
4388 assert!(
4389 !blocks.is_empty(),
4390 "Storage: empty shard at key {}",
4391 key.sharded_key.highest_block_number
4392 );
4393 }
4394
4395 if !shards.is_empty() {
4397 let last = shards.last().unwrap();
4398 assert_eq!(
4399 last.0.sharded_key.highest_block_number,
4400 u64::MAX,
4401 "Storage: last shard must be sentinel"
4402 );
4403 }
4404 }
4405 }
4406 }
4407
4408 #[test]
4409 fn test_prune_account_history_batch_multiple_sorted_targets() {
4410 let temp_dir = TempDir::new().unwrap();
4411 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4412
4413 let addr1 = Address::from([0x01; 20]);
4414 let addr2 = Address::from([0x02; 20]);
4415 let addr3 = Address::from([0x03; 20]);
4416
4417 let mut batch = provider.batch();
4419 batch
4420 .put::<tables::AccountsHistory>(
4421 ShardedKey::new(addr1, u64::MAX),
4422 &BlockNumberList::new_pre_sorted([10, 20, 30]),
4423 )
4424 .unwrap();
4425 batch
4426 .put::<tables::AccountsHistory>(
4427 ShardedKey::new(addr2, u64::MAX),
4428 &BlockNumberList::new_pre_sorted([5, 10, 15]),
4429 )
4430 .unwrap();
4431 batch
4432 .put::<tables::AccountsHistory>(
4433 ShardedKey::new(addr3, u64::MAX),
4434 &BlockNumberList::new_pre_sorted([100, 200]),
4435 )
4436 .unwrap();
4437 batch.commit().unwrap();
4438
4439 let mut targets = vec![(addr1, 15), (addr2, 10), (addr3, 50)];
4441 targets.sort_by_key(|(addr, _)| *addr);
4442
4443 let mut batch = provider.batch();
4444 let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4445 batch.commit().unwrap();
4446
4447 assert_eq!(outcomes.updated, 2);
4451 assert_eq!(outcomes.unchanged, 1);
4452
4453 let shards1 = provider.account_history_shards(addr1).unwrap();
4454 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
4455
4456 let shards2 = provider.account_history_shards(addr2).unwrap();
4457 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15]);
4458
4459 let shards3 = provider.account_history_shards(addr3).unwrap();
4460 assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![100, 200]);
4461 }
4462
4463 #[test]
4464 fn test_prune_account_history_batch_target_with_no_shards() {
4465 let temp_dir = TempDir::new().unwrap();
4466 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4467
4468 let addr1 = Address::from([0x01; 20]);
4469 let addr2 = Address::from([0x02; 20]); let addr3 = Address::from([0x03; 20]);
4471
4472 let mut batch = provider.batch();
4474 batch
4475 .put::<tables::AccountsHistory>(
4476 ShardedKey::new(addr1, u64::MAX),
4477 &BlockNumberList::new_pre_sorted([10, 20]),
4478 )
4479 .unwrap();
4480 batch
4481 .put::<tables::AccountsHistory>(
4482 ShardedKey::new(addr3, u64::MAX),
4483 &BlockNumberList::new_pre_sorted([30, 40]),
4484 )
4485 .unwrap();
4486 batch.commit().unwrap();
4487
4488 let mut targets = vec![(addr1, 15), (addr2, 100), (addr3, 35)];
4490 targets.sort_by_key(|(addr, _)| *addr);
4491
4492 let mut batch = provider.batch();
4493 let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4494 batch.commit().unwrap();
4495
4496 assert_eq!(outcomes.updated, 2);
4500 assert_eq!(outcomes.unchanged, 1);
4501
4502 let shards1 = provider.account_history_shards(addr1).unwrap();
4503 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20]);
4504
4505 let shards3 = provider.account_history_shards(addr3).unwrap();
4506 assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![40]);
4507 }
4508
4509 #[test]
4510 fn test_prune_storage_history_batch_multiple_sorted_targets() {
4511 let temp_dir = TempDir::new().unwrap();
4512 let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4513
4514 let addr = Address::from([0x42; 20]);
4515 let slot1 = B256::from([0x01; 32]);
4516 let slot2 = B256::from([0x02; 32]);
4517
4518 let mut batch = provider.batch();
4520 batch
4521 .put::<tables::StoragesHistory>(
4522 StorageShardedKey::new(addr, slot1, u64::MAX),
4523 &BlockNumberList::new_pre_sorted([10, 20, 30]),
4524 )
4525 .unwrap();
4526 batch
4527 .put::<tables::StoragesHistory>(
4528 StorageShardedKey::new(addr, slot2, u64::MAX),
4529 &BlockNumberList::new_pre_sorted([5, 15, 25]),
4530 )
4531 .unwrap();
4532 batch.commit().unwrap();
4533
4534 let mut targets = vec![((addr, slot1), 15), ((addr, slot2), 10)];
4536 targets.sort_by_key(|((a, s), _)| (*a, *s));
4537
4538 let mut batch = provider.batch();
4539 let outcomes = batch.prune_storage_history_batch(&targets).unwrap();
4540 batch.commit().unwrap();
4541
4542 assert_eq!(outcomes.updated, 2);
4543
4544 let shards1 = provider.storage_history_shards(addr, slot1).unwrap();
4545 assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
4546
4547 let shards2 = provider.storage_history_shards(addr, slot2).unwrap();
4548 assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15, 25]);
4549 }
4550}