Skip to main content

reth_provider/providers/rocksdb/
provider.rs

1use super::metrics::{RocksDBMetrics, RocksDBOperation, ROCKSDB_TABLES};
2use crate::providers::{compute_history_rank, needs_prev_shard_check, HistoryInfo};
3use alloy_consensus::transaction::TxHashRef;
4use alloy_primitives::{
5    map::{AddressMap, HashMap},
6    Address, BlockNumber, TxNumber, B256,
7};
8use itertools::Itertools;
9use metrics::Label;
10use parking_lot::Mutex;
11use rayon::prelude::*;
12use reth_chain_state::ExecutedBlock;
13use reth_db_api::{
14    database_metrics::DatabaseMetrics,
15    models::{
16        sharded_key::NUM_OF_INDICES_IN_SHARD, storage_sharded_key::StorageShardedKey, ShardedKey,
17        StorageSettings,
18    },
19    table::{Compress, Decode, Decompress, Encode, Table},
20    tables, BlockNumberList, DatabaseError,
21};
22use reth_primitives_traits::{BlockBody as _, FastInstant as Instant};
23use reth_prune_types::PruneMode;
24use reth_storage_errors::{
25    db::{DatabaseErrorInfo, DatabaseWriteError, DatabaseWriteOperation, LogLevel},
26    provider::{ProviderError, ProviderResult},
27};
28use rocksdb::{
29    statistics::StatsLevel, BlockBasedOptions, Cache, ColumnFamilyDescriptor, CompactionPri,
30    DBCompressionType, DBRawIteratorWithThreadMode, IteratorMode, OptimisticTransactionDB,
31    OptimisticTransactionOptions, Options, ReadOptions, SnapshotWithThreadMode, Transaction,
32    WriteBatchWithTransaction, WriteBufferManager, WriteOptions, DB, DEFAULT_COLUMN_FAMILY_NAME,
33};
34use std::{
35    collections::BTreeMap,
36    fmt,
37    path::{Path, PathBuf},
38    sync::Arc,
39};
40use tracing::instrument;
41
42/// Returns [`WriteOptions`] with WAL sync enabled for crash durability.
43fn synced_write_options() -> WriteOptions {
44    let mut opts = WriteOptions::default();
45    opts.set_sync(true);
46    opts
47}
48
49/// Pending `RocksDB` batches type alias.
50pub(crate) type PendingRocksDBBatches = Arc<Mutex<Vec<WriteBatchWithTransaction<true>>>>;
51
52/// Raw key-value result from a `RocksDB` iterator.
53type RawKVResult = Result<(Box<[u8]>, Box<[u8]>), rocksdb::Error>;
54
55/// Statistics for a single `RocksDB` table (column family).
56#[derive(Debug, Clone)]
57pub struct RocksDBTableStats {
58    /// Size of SST files on disk in bytes.
59    pub sst_size_bytes: u64,
60    /// Size of memtables in memory in bytes.
61    pub memtable_size_bytes: u64,
62    /// Name of the table/column family.
63    pub name: String,
64    /// Estimated number of keys in the table.
65    pub estimated_num_keys: u64,
66    /// Estimated size of live data in bytes (SST files + memtables).
67    pub estimated_size_bytes: u64,
68    /// Estimated bytes pending compaction (reclaimable space).
69    pub pending_compaction_bytes: u64,
70}
71
72/// Database-level statistics for `RocksDB`.
73///
74/// Contains both per-table statistics and DB-level metrics like WAL size.
75#[derive(Debug, Clone)]
76pub struct RocksDBStats {
77    /// Statistics for each table (column family).
78    pub tables: Vec<RocksDBTableStats>,
79    /// Total size of WAL (Write-Ahead Log) files in bytes.
80    ///
81    /// WAL is shared across all tables and not included in per-table metrics.
82    pub wal_size_bytes: u64,
83}
84
85/// Context for `RocksDB` block writes.
86#[derive(Clone)]
87pub(crate) struct RocksDBWriteCtx {
88    /// The first block number being written.
89    pub first_block_number: BlockNumber,
90    /// The prune mode for transaction lookup, if any.
91    pub prune_tx_lookup: Option<PruneMode>,
92    /// Storage settings determining what goes to `RocksDB`.
93    pub storage_settings: StorageSettings,
94    /// Pending batches to push to after writing.
95    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
109/// Default cache size for `RocksDB` block cache (128 MB).
110const DEFAULT_CACHE_SIZE: usize = 128 << 20;
111
112/// Default block size for `RocksDB` tables (16 KB).
113const DEFAULT_BLOCK_SIZE: usize = 16 * 1024;
114
115/// Default max background jobs for `RocksDB` compaction and flushing.
116const DEFAULT_MAX_BACKGROUND_JOBS: i32 = 6;
117
118/// Max open file descriptors for `RocksDB` when the process limit is low.
119///
120/// Caps the number of SST file handles `RocksDB` keeps open simultaneously.
121/// Set to 512 to stay within the common default OS `ulimit -n` of 1024,
122/// leaving headroom for MDBX, static files, and other I/O.
123const LIMITED_MAX_OPEN_FILES: i32 = 512;
124
125/// Keeps all `RocksDB` files open and avoids table cache lookups.
126const KEEP_ALL_FILES_OPEN: i32 = -1;
127
128/// Minimum process file descriptor limit for keeping all `RocksDB` files open.
129///
130/// Mature archive databases can use tens of thousands of file descriptors. This threshold keeps
131/// unlimited mode for hosts configured for that workload while retaining a bounded table cache on
132/// systems with low limits.
133const HIGH_FILE_DESCRIPTOR_LIMIT: u64 = 128 * 1024;
134
135/// Default bytes per sync for `RocksDB` WAL writes (1 MB).
136const DEFAULT_BYTES_PER_SYNC: u64 = 1_048_576;
137
138/// Default write buffer size for `RocksDB` memtables (128 MB).
139///
140/// Larger memtables reduce flush frequency during burst writes, providing more consistent
141/// tail latency. Benchmarks showed 128 MB reduces p99 latency variance by ~80% compared
142/// to 64 MB default, with negligible impact on mean throughput.
143const DEFAULT_WRITE_BUFFER_SIZE: usize = 128 << 20;
144
145/// Default total `RocksDB` memtable memory budget across column families (4 GiB).
146///
147/// This is a soft limit; with write stalls enabled, `RocksDB` waits for flushes once
148/// memtable arena usage exceeds the budget.
149const DEFAULT_WRITE_BUFFER_MANAGER_SIZE: usize = 4 * 1024 * 1024 * 1024;
150
151/// Default buffer capacity for compression in batches.
152/// 4 KiB matches common block/page sizes and comfortably holds typical history values,
153/// reducing the first few reallocations without over-allocating.
154const DEFAULT_COMPRESS_BUF_CAPACITY: usize = 4096;
155
156/// Default auto-commit threshold for batch writes (512 MiB).
157///
158/// When a batch exceeds this size, it is automatically committed to prevent OOM
159/// during large bulk writes. Keep this below the `RocksDB` write buffer manager
160/// budget so stalls can recover without waiting on a single large flush.
161/// The consistency check on startup heals any crash that occurs between auto-commits.
162const DEFAULT_AUTO_COMMIT_THRESHOLD: usize = 512 * 1024 * 1024;
163
164/// Minimum BAL value size stored in `BlobDB` files.
165///
166/// Smaller BALs stay inline. Larger payloads avoid regular LSM value compaction.
167const DEFAULT_BAL_MIN_BLOB_SIZE: u64 = 4 * 1024;
168
169/// Target BAL blob file size.
170const DEFAULT_BAL_BLOB_FILE_SIZE: u64 = 256 * 1024 * 1024;
171
172/// Builder for [`RocksDBProvider`].
173pub struct RocksDBBuilder {
174    path: PathBuf,
175    column_families: Vec<String>,
176    enable_metrics: bool,
177    enable_statistics: bool,
178    log_level: rocksdb::LogLevel,
179    block_cache: Cache,
180    read_only: bool,
181}
182
183impl fmt::Debug for RocksDBBuilder {
184    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
185        f.debug_struct("RocksDBBuilder")
186            .field("path", &self.path)
187            .field("column_families", &self.column_families)
188            .field("enable_metrics", &self.enable_metrics)
189            .finish()
190    }
191}
192
193impl RocksDBBuilder {
194    /// Creates a new builder with optimized default options.
195    pub fn new(path: impl AsRef<Path>) -> Self {
196        let cache = Cache::new_lru_cache(DEFAULT_CACHE_SIZE);
197        Self {
198            path: path.as_ref().to_path_buf(),
199            column_families: Vec::new(),
200            enable_metrics: false,
201            enable_statistics: false,
202            log_level: rocksdb::LogLevel::Info,
203            block_cache: cache,
204            read_only: false,
205        }
206    }
207
208    /// Creates default table options with shared block cache.
209    fn default_table_options(cache: &Cache) -> BlockBasedOptions {
210        let mut table_options = BlockBasedOptions::default();
211        table_options.set_block_size(DEFAULT_BLOCK_SIZE);
212        table_options.set_cache_index_and_filter_blocks(true);
213        table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
214        // Shared block cache for all column families.
215        table_options.set_block_cache(cache);
216        table_options
217    }
218
219    /// Creates optimized `RocksDB` options per `RocksDB` wiki recommendations.
220    fn default_options(
221        log_level: rocksdb::LogLevel,
222        cache: &Cache,
223        enable_statistics: bool,
224    ) -> Options {
225        // Follow recommend tuning guide from RocksDB wiki, see https://github.com/facebook/rocksdb/wiki/Setup-Options-and-Basic-Tuning
226        let table_options = Self::default_table_options(cache);
227
228        let mut options = Options::default();
229        options.set_block_based_table_factory(&table_options);
230        options.create_if_missing(true);
231        options.create_missing_column_families(true);
232        options.set_max_background_jobs(DEFAULT_MAX_BACKGROUND_JOBS);
233        options.set_bytes_per_sync(DEFAULT_BYTES_PER_SYNC);
234        let write_buffer_manager =
235            WriteBufferManager::new_write_buffer_manager(DEFAULT_WRITE_BUFFER_MANAGER_SIZE, true);
236        options.set_write_buffer_manager(&write_buffer_manager);
237
238        options.set_bottommost_compression_type(DBCompressionType::Zstd);
239        options.set_bottommost_zstd_max_train_bytes(0, true);
240        options.set_compression_type(DBCompressionType::Lz4);
241        options.set_compaction_pri(CompactionPri::MinOverlappingRatio);
242
243        options.set_log_level(log_level);
244
245        options.set_max_open_files(select_max_open_files());
246
247        // Delete obsolete WAL files immediately after all column families have flushed.
248        // Both set to 0 means "delete ASAP, no archival".
249        options.set_wal_ttl_seconds(0);
250        options.set_wal_size_limit_mb(0);
251
252        // Statistics can view from RocksDB log file
253        if enable_statistics {
254            options.enable_statistics();
255            // Nothing reads the statistics programmatically, they only end up in the periodic
256            // LOG dump, so collect the counters but not the timer histograms: at the default
257            // level every `Get`, `Seek` and write wraps itself in a `StopWatch` that reads the
258            // clock twice, while the tickers are plain atomic increments.
259            //
260            // Note: the discriminants of `rocksdb::StatsLevel` are one higher than the levels the
261            // C API defines, so this variant selects "except timers". Should that ever be
262            // corrected upstream it selects "except histogram or timers", which also keeps the
263            // tickers and drops the timers.
264            options.set_statistics_level(StatsLevel::ExceptHistogramOrTimers);
265        }
266
267        options
268    }
269
270    /// Creates optimized column family options.
271    fn default_column_family_options(cache: &Cache) -> Options {
272        // Follow recommend tuning guide from RocksDB wiki, see https://github.com/facebook/rocksdb/wiki/Setup-Options-and-Basic-Tuning
273        let table_options = Self::default_table_options(cache);
274
275        let mut cf_options = Options::default();
276        cf_options.set_block_based_table_factory(&table_options);
277        cf_options.set_level_compaction_dynamic_level_bytes(true);
278        // Recommend to use Zstd for bottommost compression and Lz4 for other levels, see https://github.com/facebook/rocksdb/wiki/Compression#configuration
279        cf_options.set_compression_type(DBCompressionType::Lz4);
280        cf_options.set_bottommost_compression_type(DBCompressionType::Zstd);
281        // Only use Zstd compression, disable dictionary training
282        cf_options.set_bottommost_zstd_max_train_bytes(0, true);
283        cf_options.set_write_buffer_size(DEFAULT_WRITE_BUFFER_SIZE);
284
285        cf_options
286    }
287
288    /// Creates column family options for block access list payloads.
289    fn block_access_lists_column_family_options(cache: &Cache) -> Options {
290        let mut cf_options = Self::default_column_family_options(cache);
291        cf_options.set_enable_blob_files(true);
292        cf_options.set_min_blob_size(DEFAULT_BAL_MIN_BLOB_SIZE);
293        cf_options.set_blob_file_size(DEFAULT_BAL_BLOB_FILE_SIZE);
294        cf_options.set_blob_compression_type(DBCompressionType::Lz4);
295        cf_options
296    }
297
298    /// Creates optimized column family options for `TransactionHashNumbers`.
299    ///
300    /// This table stores `B256 -> TxNumber` mappings where:
301    /// - Keys are incompressible 32-byte hashes (compression wastes CPU for zero benefit)
302    /// - Values are varint-encoded `u64` (a few bytes - too small to benefit from compression)
303    /// - Every lookup expects a hit (bloom filters only help when checking non-existent keys)
304    fn tx_hash_numbers_column_family_options(cache: &Cache) -> Options {
305        let mut table_options = BlockBasedOptions::default();
306        table_options.set_block_size(DEFAULT_BLOCK_SIZE);
307        table_options.set_cache_index_and_filter_blocks(true);
308        table_options.set_pin_l0_filter_and_index_blocks_in_cache(true);
309        table_options.set_block_cache(cache);
310        // Disable bloom filter: every lookup expects a hit, so bloom filters provide no benefit
311        // and waste memory
312
313        let mut cf_options = Options::default();
314        cf_options.set_block_based_table_factory(&table_options);
315        cf_options.set_level_compaction_dynamic_level_bytes(true);
316        // Disable compression: B256 keys are incompressible hashes, TxNumber values are
317        // varint-encoded u64 (a few bytes). Compression wastes CPU cycles for zero space savings.
318        cf_options.set_compression_type(DBCompressionType::None);
319        cf_options.set_bottommost_compression_type(DBCompressionType::None);
320
321        cf_options
322    }
323
324    /// Adds a column family for a specific table type.
325    pub fn with_table<T: Table>(mut self) -> Self {
326        self.column_families.push(T::NAME.to_string());
327        self
328    }
329
330    /// Registers the default tables used by reth for `RocksDB` storage.
331    ///
332    /// This registers:
333    /// - [`tables::TransactionHashNumbers`] - Transaction hash to number mapping
334    /// - [`tables::AccountsHistory`] - Account history index
335    /// - [`tables::StoragesHistory`] - Storage history index
336    /// - [`tables::BlockAccessLists`] - Block access list payloads
337    /// - [`tables::BlockAccessListBlockNumbers`] - Block access list hash index
338    pub fn with_default_tables(self) -> Self {
339        self.with_table::<tables::TransactionHashNumbers>()
340            .with_table::<tables::AccountsHistory>()
341            .with_table::<tables::StoragesHistory>()
342            .with_table::<tables::BlockAccessLists>()
343            .with_table::<tables::BlockAccessListBlockNumbers>()
344    }
345
346    /// Enables metrics.
347    pub const fn with_metrics(mut self) -> Self {
348        self.enable_metrics = true;
349        self
350    }
351
352    /// Enables `RocksDB` internal statistics collection.
353    pub const fn with_statistics(mut self) -> Self {
354        self.enable_statistics = true;
355        self
356    }
357
358    /// Sets the log level from `DatabaseArgs` configuration.
359    pub const fn with_database_log_level(mut self, log_level: Option<LogLevel>) -> Self {
360        if let Some(level) = log_level {
361            self.log_level = convert_log_level(level);
362        }
363        self
364    }
365
366    /// Sets a custom block cache size.
367    pub fn with_block_cache_size(mut self, capacity_bytes: usize) -> Self {
368        self.block_cache = Cache::new_lru_cache(capacity_bytes);
369        self
370    }
371
372    /// Sets a custom block cache size if provided, otherwise keeps the current cache.
373    pub fn with_block_cache_size_opt(self, capacity_bytes: Option<usize>) -> Self {
374        if let Some(capacity_bytes) = capacity_bytes {
375            self.with_block_cache_size(capacity_bytes)
376        } else {
377            self
378        }
379    }
380
381    /// Sets read-only mode.
382    ///
383    /// Opens the database as a secondary instance, which supports catching up
384    /// with the primary via [`RocksDBProvider::try_catch_up_with_primary`].
385    /// A temporary directory is created automatically for the secondary's LOG files.
386    ///
387    /// Note: Write operations on a read-only provider will panic at runtime.
388    pub const fn with_read_only(mut self, read_only: bool) -> Self {
389        self.read_only = read_only;
390        self
391    }
392
393    /// Builds the [`RocksDBProvider`].
394    pub fn build(self) -> ProviderResult<RocksDBProvider> {
395        let options =
396            Self::default_options(self.log_level, &self.block_cache, self.enable_statistics);
397
398        let mut cf_descriptors: Vec<ColumnFamilyDescriptor> = self
399            .column_families
400            .iter()
401            .map(|name| {
402                let cf_options = if name == tables::TransactionHashNumbers::NAME {
403                    Self::tx_hash_numbers_column_family_options(&self.block_cache)
404                } else if name == tables::BlockAccessLists::NAME {
405                    Self::block_access_lists_column_family_options(&self.block_cache)
406                } else {
407                    Self::default_column_family_options(&self.block_cache)
408                };
409                ColumnFamilyDescriptor::new(name.clone(), cf_options)
410            })
411            .collect();
412
413        // RocksDB requires every existing column family to be opened. Preserve column families
414        // unknown to this configuration so databases remain openable after a downgrade.
415        if RocksDBProvider::exists(&self.path) {
416            let existing_column_families = DB::list_cf(&options, &self.path).map_err(|e| {
417                ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
418                    message: e.to_string().into(),
419                    code: -1,
420                }))
421            })?;
422            if self.read_only {
423                // Legacy databases have no BAL tables; secondary opens cannot create them.
424                // Keep existing BAL tables; only omit those absent from disk.
425                cf_descriptors.retain(|cf| {
426                    !matches!(
427                        cf.name(),
428                        tables::BlockAccessLists::NAME | tables::BlockAccessListBlockNumbers::NAME
429                    ) || existing_column_families.iter().any(|name| name == cf.name())
430                });
431            }
432            let unknown_column_families: Vec<String> = existing_column_families
433                .into_iter()
434                .filter(|name| {
435                    name != DEFAULT_COLUMN_FAMILY_NAME && !self.column_families.contains(name)
436                })
437                .collect();
438            if !unknown_column_families.is_empty() {
439                tracing::debug!(
440                    target: "providers::rocksdb",
441                    column_families = ?unknown_column_families,
442                    "Preserving unknown column families"
443                );
444                cf_descriptors.extend(unknown_column_families.into_iter().map(|name| {
445                    ColumnFamilyDescriptor::new(
446                        name,
447                        Self::default_column_family_options(&self.block_cache),
448                    )
449                }));
450            }
451        }
452
453        let metrics = self.enable_metrics.then(RocksDBMetrics::default);
454
455        if self.read_only {
456            // Open as secondary instance for catch-up capability.
457            // Secondary needs max_open_files = -1 to keep all FDs open.
458            let mut options = options;
459            options.set_max_open_files(KEEP_ALL_FILES_OPEN);
460
461            let secondary_path = self
462                .path
463                .parent()
464                .unwrap_or(&self.path)
465                .join(format!("rocksdb-secondary-tmp-{}", std::process::id()));
466            reth_fs_util::create_dir_all(&secondary_path).map_err(ProviderError::other)?;
467
468            let db = DB::open_cf_descriptors_as_secondary(
469                &options,
470                &self.path,
471                &secondary_path,
472                cf_descriptors,
473            )
474            .map_err(|e| {
475                ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
476                    message: e.to_string().into(),
477                    code: -1,
478                }))
479            })?;
480            Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::Secondary {
481                db,
482                metrics,
483                secondary_path,
484            })))
485        } else {
486            // Use OptimisticTransactionDB for MDBX-like transaction semantics (read-your-writes,
487            // rollback) OptimisticTransactionDB uses optimistic concurrency control (conflict
488            // detection at commit) and is backed by DBCommon, giving us access to
489            // cancel_all_background_work for clean shutdown.
490            let db =
491                OptimisticTransactionDB::open_cf_descriptors(&options, &self.path, cf_descriptors)
492                    .map_err(|e| {
493                        ProviderError::Database(DatabaseError::Open(DatabaseErrorInfo {
494                            message: e.to_string().into(),
495                            code: -1,
496                        }))
497                    })?;
498            Ok(RocksDBProvider(Arc::new(RocksDBProviderInner::ReadWrite { db, metrics })))
499        }
500    }
501}
502
503/// Some types don't support compression (eg. B256), and we don't want to be copying them to the
504/// allocated buffer when we can just use their reference.
505macro_rules! compress_to_buf_or_ref {
506    ($buf:expr, $value:expr) => {
507        if let Some(value) = $value.uncompressable_ref() {
508            Some(value)
509        } else {
510            $buf.clear();
511            $value.compress_to_buf(&mut $buf);
512            None
513        }
514    };
515}
516
517/// `RocksDB` provider for auxiliary storage layer beside main database MDBX.
518#[derive(Debug)]
519pub struct RocksDBProvider(Arc<RocksDBProviderInner>);
520
521/// Inner state for `RocksDB` provider.
522enum RocksDBProviderInner {
523    /// Read-write mode using `OptimisticTransactionDB`.
524    ReadWrite {
525        /// `RocksDB` database instance with optimistic transaction support.
526        db: OptimisticTransactionDB,
527        /// Metrics latency & operations.
528        metrics: Option<RocksDBMetrics>,
529    },
530    /// Secondary mode using `DB` opened with `open_cf_descriptors_as_secondary`.
531    /// Supports catching up with the primary via `try_catch_up_with_primary`.
532    /// Does not support snapshots; consistency is guaranteed externally.
533    Secondary {
534        /// Secondary `RocksDB` database instance.
535        db: DB,
536        /// Metrics latency & operations.
537        metrics: Option<RocksDBMetrics>,
538        /// Temporary directory for secondary LOG files, removed on drop.
539        secondary_path: PathBuf,
540    },
541}
542
543impl RocksDBProviderInner {
544    /// Returns the metrics for this provider.
545    const fn metrics(&self) -> Option<&RocksDBMetrics> {
546        match self {
547            Self::ReadWrite { metrics, .. } | Self::Secondary { metrics, .. } => metrics.as_ref(),
548        }
549    }
550
551    /// Returns the read-write database, panicking if in read-only mode.
552    fn db_rw(&self) -> &OptimisticTransactionDB {
553        match self {
554            Self::ReadWrite { db, .. } => db,
555            Self::Secondary { .. } => {
556                panic!("Cannot perform write operation on secondary RocksDB provider")
557            }
558        }
559    }
560
561    /// Gets the column family handle for a table.
562    fn cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
563        let cf = match self {
564            Self::ReadWrite { db, .. } => db.cf_handle(T::NAME),
565            Self::Secondary { db, .. } => db.cf_handle(T::NAME),
566        };
567        cf.ok_or_else(|| DatabaseError::Other(format!("Column family '{}' not found", T::NAME)))
568    }
569
570    /// Gets a value from a column family.
571    fn get_cf(
572        &self,
573        cf: &rocksdb::ColumnFamily,
574        key: impl AsRef<[u8]>,
575    ) -> Result<Option<Vec<u8>>, rocksdb::Error> {
576        match self {
577            Self::ReadWrite { db, .. } => db.get_cf(cf, key),
578            Self::Secondary { db, .. } => db.get_cf(cf, key),
579        }
580    }
581
582    /// Puts a value into a column family.
583    fn put_cf(
584        &self,
585        cf: &rocksdb::ColumnFamily,
586        key: impl AsRef<[u8]>,
587        value: impl AsRef<[u8]>,
588    ) -> Result<(), rocksdb::Error> {
589        self.db_rw().put_cf(cf, key, value)
590    }
591
592    /// Deletes a value from a column family.
593    fn delete_cf(
594        &self,
595        cf: &rocksdb::ColumnFamily,
596        key: impl AsRef<[u8]>,
597    ) -> Result<(), rocksdb::Error> {
598        self.db_rw().delete_cf(cf, key)
599    }
600
601    /// Deletes a range of values from a column family.
602    fn delete_range_cf<K: AsRef<[u8]>>(
603        &self,
604        cf: &rocksdb::ColumnFamily,
605        from: K,
606        to: K,
607    ) -> Result<(), rocksdb::Error> {
608        self.db_rw().delete_range_cf(cf, from, to)
609    }
610
611    /// Returns an iterator over a column family.
612    fn iterator_cf(
613        &self,
614        cf: &rocksdb::ColumnFamily,
615        mode: IteratorMode<'_>,
616    ) -> RocksDBIterEnum<'_> {
617        match self {
618            Self::ReadWrite { db, .. } => RocksDBIterEnum::ReadWrite(db.iterator_cf(cf, mode)),
619            Self::Secondary { db, .. } => RocksDBIterEnum::ReadOnly(db.iterator_cf(cf, mode)),
620        }
621    }
622
623    /// Returns a raw iterator over a column family.
624    ///
625    /// Unlike [`Self::iterator_cf`], raw iterators support `seek()` for efficient
626    /// repositioning without creating a new iterator.
627    fn raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'_> {
628        match self {
629            Self::ReadWrite { db, .. } => RocksDBRawIterEnum::ReadWrite(db.raw_iterator_cf(cf)),
630            Self::Secondary { db, .. } => RocksDBRawIterEnum::ReadOnly(db.raw_iterator_cf(cf)),
631        }
632    }
633
634    /// Returns a forward iterator over a column family over the keys in `[from, last]`.
635    ///
636    /// The bound matters when the range is followed by many deleted keys: an unbounded iterator
637    /// that runs out of live keys in the range skips every tombstone after it until it finds the
638    /// next live key, whereas a bounded one stops at the bound.
639    fn iterator_cf_inclusive_range(
640        &self,
641        cf: &rocksdb::ColumnFamily,
642        from: &[u8],
643        last: &[u8],
644    ) -> RocksDBIterEnum<'_> {
645        // The smallest key greater than `last` is `last` followed by a zero byte.
646        let mut upper_bound = Vec::with_capacity(last.len() + 1);
647        upper_bound.extend_from_slice(last);
648        upper_bound.push(0);
649
650        let mut opts = ReadOptions::default();
651        opts.set_iterate_upper_bound(upper_bound);
652        let mode = IteratorMode::From(from, rocksdb::Direction::Forward);
653        match self {
654            Self::ReadWrite { db, .. } => {
655                RocksDBIterEnum::ReadWrite(db.iterator_cf_opt(cf, opts, mode))
656            }
657            Self::Secondary { db, .. } => {
658                RocksDBIterEnum::ReadOnly(db.iterator_cf_opt(cf, opts, mode))
659            }
660        }
661    }
662
663    /// Returns a read-only, point-in-time snapshot of the database.
664    fn snapshot(&self) -> RocksReadSnapshotInner<'_> {
665        match self {
666            Self::ReadWrite { db, .. } => RocksReadSnapshotInner::ReadWrite(db.snapshot()),
667            Self::Secondary { db, .. } => RocksReadSnapshotInner::Secondary(db),
668        }
669    }
670
671    /// Returns the path to the database directory.
672    fn path(&self) -> &Path {
673        match self {
674            Self::ReadWrite { db, .. } => db.path(),
675            Self::Secondary { db, .. } => db.path(),
676        }
677    }
678
679    /// Returns the total size of WAL (Write-Ahead Log) files in bytes.
680    ///
681    /// WAL files have a `.log` extension in the `RocksDB` directory.
682    fn wal_size_bytes(&self) -> u64 {
683        let path = self.path();
684
685        match std::fs::read_dir(path) {
686            Ok(entries) => entries
687                .filter_map(|e| e.ok())
688                .filter(|e| e.path().extension().is_some_and(|ext| ext == "log"))
689                .filter_map(|e| e.metadata().ok())
690                .map(|m| m.len())
691                .sum(),
692            Err(_) => 0,
693        }
694    }
695
696    /// Returns statistics for all column families in the database.
697    fn table_stats(&self) -> Vec<RocksDBTableStats> {
698        let mut stats = Vec::new();
699
700        macro_rules! collect_stats {
701            ($db:expr) => {
702                for cf_name in ROCKSDB_TABLES {
703                    if let Some(cf) = $db.cf_handle(cf_name) {
704                        let estimated_num_keys = $db
705                            .property_int_value_cf(cf, rocksdb::properties::ESTIMATE_NUM_KEYS)
706                            .ok()
707                            .flatten()
708                            .unwrap_or(0);
709
710                        // SST files size (on-disk) + memtable size (in-memory)
711                        let sst_size = $db
712                            .property_int_value_cf(cf, rocksdb::properties::LIVE_SST_FILES_SIZE)
713                            .ok()
714                            .flatten()
715                            .unwrap_or(0);
716
717                        let memtable_size = $db
718                            .property_int_value_cf(cf, rocksdb::properties::SIZE_ALL_MEM_TABLES)
719                            .ok()
720                            .flatten()
721                            .unwrap_or(0);
722
723                        let estimated_size_bytes = sst_size + memtable_size;
724
725                        let pending_compaction_bytes = $db
726                            .property_int_value_cf(
727                                cf,
728                                rocksdb::properties::ESTIMATE_PENDING_COMPACTION_BYTES,
729                            )
730                            .ok()
731                            .flatten()
732                            .unwrap_or(0);
733
734                        stats.push(RocksDBTableStats {
735                            sst_size_bytes: sst_size,
736                            memtable_size_bytes: memtable_size,
737                            name: cf_name.to_string(),
738                            estimated_num_keys,
739                            estimated_size_bytes,
740                            pending_compaction_bytes,
741                        });
742                    }
743                }
744            };
745        }
746
747        match self {
748            Self::ReadWrite { db, .. } => collect_stats!(db),
749            Self::Secondary { db, .. } => collect_stats!(db),
750        }
751
752        stats
753    }
754
755    /// Returns database-level statistics including per-table stats and WAL size.
756    fn db_stats(&self) -> RocksDBStats {
757        RocksDBStats { tables: self.table_stats(), wal_size_bytes: self.wal_size_bytes() }
758    }
759}
760
761impl fmt::Debug for RocksDBProviderInner {
762    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
763        match self {
764            Self::ReadWrite { metrics, .. } => f
765                .debug_struct("RocksDBProviderInner::ReadWrite")
766                .field("db", &"<OptimisticTransactionDB>")
767                .field("metrics", metrics)
768                .finish(),
769            Self::Secondary { metrics, .. } => f
770                .debug_struct("RocksDBProviderInner::Secondary")
771                .field("db", &"<DB (secondary)>")
772                .field("metrics", metrics)
773                .finish(),
774        }
775    }
776}
777
778impl Drop for RocksDBProviderInner {
779    fn drop(&mut self) {
780        match self {
781            Self::ReadWrite { db, .. } => {
782                // Flush all memtables if possible. If not, they will be rebuilt from the WAL on
783                // restart
784                if let Err(e) = db.flush_wal(true) {
785                    tracing::warn!(target: "providers::rocksdb", ?e, "Failed to flush WAL on drop");
786                }
787                for cf_name in ROCKSDB_TABLES {
788                    if let Some(cf) = db.cf_handle(cf_name) &&
789                        let Err(e) = db.flush_cf(&cf)
790                    {
791                        tracing::warn!(target: "providers::rocksdb", cf = cf_name, ?e, "Failed to flush CF on drop");
792                    }
793                }
794                db.cancel_all_background_work(true);
795            }
796            Self::Secondary { db, secondary_path, .. } => {
797                db.cancel_all_background_work(true);
798                let _ = std::fs::remove_dir_all(secondary_path);
799            }
800        }
801    }
802}
803
804impl Clone for RocksDBProvider {
805    fn clone(&self) -> Self {
806        Self(self.0.clone())
807    }
808}
809
810impl DatabaseMetrics for RocksDBProvider {
811    fn gauge_metrics(&self) -> Vec<(&'static str, f64, Vec<Label>)> {
812        let mut metrics = Vec::new();
813
814        for stat in self.table_stats() {
815            metrics.push((
816                "rocksdb.table_size",
817                stat.estimated_size_bytes as f64,
818                vec![Label::new("table", stat.name.clone())],
819            ));
820            metrics.push((
821                "rocksdb.table_entries",
822                stat.estimated_num_keys as f64,
823                vec![Label::new("table", stat.name.clone())],
824            ));
825            metrics.push((
826                "rocksdb.pending_compaction_bytes",
827                stat.pending_compaction_bytes as f64,
828                vec![Label::new("table", stat.name.clone())],
829            ));
830            metrics.push((
831                "rocksdb.sst_size",
832                stat.sst_size_bytes as f64,
833                vec![Label::new("table", stat.name.clone())],
834            ));
835            metrics.push((
836                "rocksdb.memtable_size",
837                stat.memtable_size_bytes as f64,
838                vec![Label::new("table", stat.name)],
839            ));
840        }
841
842        // WAL size (DB-level, shared across all tables)
843        metrics.push(("rocksdb.wal_size", self.wal_size_bytes() as f64, vec![]));
844
845        metrics
846    }
847}
848
849impl RocksDBProvider {
850    /// Creates a new `RocksDB` provider.
851    pub fn new(path: impl AsRef<Path>) -> ProviderResult<Self> {
852        RocksDBBuilder::new(path).build()
853    }
854
855    /// Creates a new `RocksDB` provider builder.
856    pub fn builder(path: impl AsRef<Path>) -> RocksDBBuilder {
857        RocksDBBuilder::new(path)
858    }
859
860    /// Returns `true` if a `RocksDB` database exists at the given path.
861    ///
862    /// Checks for the presence of the `CURRENT` file, which `RocksDB` creates
863    /// when initializing a database.
864    pub fn exists(path: impl AsRef<Path>) -> bool {
865        path.as_ref().join("CURRENT").exists()
866    }
867
868    /// Returns `true` if this provider is in read-only mode.
869    pub fn is_read_only(&self) -> bool {
870        matches!(self.0.as_ref(), RocksDBProviderInner::Secondary { .. })
871    }
872
873    /// Tries to catch up with the primary instance by reading new WAL and MANIFEST entries.
874    ///
875    /// This is a no-op for read-write and read-only providers.
876    /// For secondary providers, this incrementally syncs with the primary's latest state.
877    pub fn try_catch_up_with_primary(&self) -> ProviderResult<()> {
878        match self.0.as_ref() {
879            RocksDBProviderInner::Secondary { db, .. } => {
880                db.try_catch_up_with_primary().map_err(|e| {
881                    ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
882                        message: e.to_string().into(),
883                        code: -1,
884                    }))
885                })
886            }
887            _ => Ok(()),
888        }
889    }
890
891    /// Returns a read-only, point-in-time snapshot of the database.
892    ///
893    /// Lighter weight than [`RocksTx`] — no write-conflict tracking, and `Send + Sync`.
894    pub fn snapshot(&self) -> RocksReadSnapshot<'_> {
895        RocksReadSnapshot {
896            accounts_history_iter: Mutex::new(None),
897            storages_history_iter: Mutex::new(None),
898            inner: self.0.snapshot(),
899            provider: &self.0,
900        }
901    }
902
903    /// Returns a read-only, point-in-time snapshot that owns the provider handle it reads through.
904    ///
905    /// Lets a reader keep one snapshot (and the iterators cached in it) alive for its whole
906    /// lifetime, instead of creating a snapshot per lookup.
907    pub fn owned_snapshot(&self) -> OwnedRocksReadSnapshot {
908        OwnedRocksReadSnapshot::new(self)
909    }
910
911    /// Creates a new transaction with MDBX-like semantics (read-your-writes, rollback).
912    ///
913    /// Note: With `OptimisticTransactionDB`, commits may fail if there are conflicts.
914    /// Conflict detection happens at commit time, not at write time.
915    ///
916    /// # Panics
917    /// Panics if the provider is in read-only mode.
918    pub fn tx(&self) -> RocksTx<'_> {
919        let write_options = synced_write_options();
920        let txn_options = OptimisticTransactionOptions::default();
921        let inner = self.0.db_rw().transaction_opt(&write_options, &txn_options);
922        RocksTx { inner, provider: self }
923    }
924
925    /// Creates a new batch for atomic writes.
926    ///
927    /// Use [`Self::write_batch`] for closure-based atomic writes.
928    /// Use this method when the batch needs to be held by [`crate::EitherWriter`].
929    ///
930    /// # Panics
931    /// Panics if the provider is in read-only mode when attempting to commit.
932    pub fn batch(&self) -> RocksDBBatch<'_> {
933        RocksDBBatch {
934            provider: self,
935            inner: WriteBatchWithTransaction::<true>::default(),
936            buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
937            auto_commit_threshold: None,
938        }
939    }
940
941    /// Creates a new batch with auto-commit enabled.
942    ///
943    /// When the batch size exceeds the threshold (4 GiB), the batch is automatically
944    /// committed and reset. This prevents OOM during large bulk writes while maintaining
945    /// crash-safety via the consistency check on startup.
946    pub fn batch_with_auto_commit(&self) -> RocksDBBatch<'_> {
947        RocksDBBatch {
948            provider: self,
949            inner: WriteBatchWithTransaction::<true>::default(),
950            buf: Vec::with_capacity(DEFAULT_COMPRESS_BUF_CAPACITY),
951            auto_commit_threshold: Some(DEFAULT_AUTO_COMMIT_THRESHOLD),
952        }
953    }
954
955    /// Gets the column family handle for a table.
956    fn get_cf_handle<T: Table>(&self) -> Result<&rocksdb::ColumnFamily, DatabaseError> {
957        self.0.cf_handle::<T>()
958    }
959
960    /// Returns whether this provider opened the given table.
961    pub(crate) fn has_table<T: Table>(&self) -> bool {
962        match self.0.as_ref() {
963            RocksDBProviderInner::ReadWrite { db, .. } => db.cf_handle(T::NAME).is_some(),
964            RocksDBProviderInner::Secondary { db, .. } => db.cf_handle(T::NAME).is_some(),
965        }
966    }
967
968    /// Executes a function and records metrics with the given operation and table name.
969    fn execute_with_operation_metric<R>(
970        &self,
971        operation: RocksDBOperation,
972        table: &'static str,
973        f: impl FnOnce(&Self) -> R,
974    ) -> R {
975        let start = self.0.metrics().map(|_| Instant::now());
976        let res = f(self);
977
978        if let (Some(start), Some(metrics)) = (start, self.0.metrics()) {
979            metrics.record_operation(operation, table, start.elapsed());
980        }
981
982        res
983    }
984
985    /// Gets a value from the specified table.
986    pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
987        self.get_encoded::<T>(&key.encode())
988    }
989
990    /// Gets a value from the specified table using pre-encoded key.
991    pub fn get_encoded<T: Table>(
992        &self,
993        key: &<T::Key as Encode>::Encoded,
994    ) -> ProviderResult<Option<T::Value>> {
995        self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
996            let result = this.0.get_cf(this.get_cf_handle::<T>()?, key.as_ref()).map_err(|e| {
997                ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
998                    message: e.to_string().into(),
999                    code: -1,
1000                }))
1001            })?;
1002
1003            Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
1004        })
1005    }
1006
1007    /// Gets raw bytes from the specified table without decompressing.
1008    pub fn get_raw<T: Table>(&self, key: T::Key) -> ProviderResult<Option<Vec<u8>>> {
1009        let encoded = key.encode();
1010        self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
1011            this.0.get_cf(this.get_cf_handle::<T>()?, encoded.as_ref()).map_err(|e| {
1012                ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1013                    message: e.to_string().into(),
1014                    code: -1,
1015                }))
1016            })
1017        })
1018    }
1019
1020    /// Puts upsert a value into the specified table with the given key.
1021    ///
1022    /// # Panics
1023    /// Panics if the provider is in read-only mode.
1024    pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
1025        let encoded_key = key.encode();
1026        self.put_encoded::<T>(&encoded_key, value)
1027    }
1028
1029    /// Puts a value into the specified table using pre-encoded key.
1030    ///
1031    /// # Panics
1032    /// Panics if the provider is in read-only mode.
1033    pub fn put_encoded<T: Table>(
1034        &self,
1035        key: &<T::Key as Encode>::Encoded,
1036        value: &T::Value,
1037    ) -> ProviderResult<()> {
1038        self.execute_with_operation_metric(RocksDBOperation::Put, T::NAME, |this| {
1039            // for simplify the code, we need allocate buf here each time because `RocksDBProvider`
1040            // is thread safe if user want to avoid allocate buf each time, they can use
1041            // write_batch api
1042            let mut buf = Vec::new();
1043            let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
1044
1045            this.0.put_cf(this.get_cf_handle::<T>()?, key, value_bytes).map_err(|e| {
1046                ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1047                    info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1048                    operation: DatabaseWriteOperation::PutUpsert,
1049                    table_name: T::NAME,
1050                    key: key.as_ref().to_vec(),
1051                })))
1052            })
1053        })
1054    }
1055
1056    /// Deletes a value from the specified table.
1057    ///
1058    /// # Panics
1059    /// Panics if the provider is in read-only mode.
1060    pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
1061        self.execute_with_operation_metric(RocksDBOperation::Delete, T::NAME, |this| {
1062            this.0.delete_cf(this.get_cf_handle::<T>()?, key.encode().as_ref()).map_err(|e| {
1063                ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
1064                    message: e.to_string().into(),
1065                    code: -1,
1066                }))
1067            })
1068        })
1069    }
1070
1071    /// Clears all entries from the specified table.
1072    ///
1073    /// Uses `delete_range_cf` from empty key to a max key (256 bytes of 0xFF).
1074    /// This end key must exceed the maximum encoded key size for any table.
1075    /// Current max is ~60 bytes (`StorageShardedKey` = 20 + 32 + 8).
1076    pub fn clear<T: Table>(&self) -> ProviderResult<()> {
1077        let cf = self.get_cf_handle::<T>()?;
1078
1079        self.0.delete_range_cf(cf, &[] as &[u8], &[0xFF; 256]).map_err(|e| {
1080            ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
1081                message: e.to_string().into(),
1082                code: -1,
1083            }))
1084        })?;
1085
1086        Ok(())
1087    }
1088
1089    /// Retrieves the first or last entry from a table based on the iterator mode.
1090    fn get_boundary<T: Table>(
1091        &self,
1092        mode: IteratorMode<'_>,
1093    ) -> ProviderResult<Option<(T::Key, T::Value)>> {
1094        self.execute_with_operation_metric(RocksDBOperation::Get, T::NAME, |this| {
1095            let cf = this.get_cf_handle::<T>()?;
1096            let mut iter = this.0.iterator_cf(cf, mode);
1097
1098            match iter.next() {
1099                Some(Ok((key_bytes, value_bytes))) => {
1100                    let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
1101                        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1102                    let value = T::Value::decompress(&value_bytes)
1103                        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1104                    Ok(Some((key, value)))
1105                }
1106                Some(Err(e)) => {
1107                    Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1108                        message: e.to_string().into(),
1109                        code: -1,
1110                    })))
1111                }
1112                None => Ok(None),
1113            }
1114        })
1115    }
1116
1117    /// Gets the first (smallest key) entry from the specified table.
1118    #[inline]
1119    pub fn first<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1120        self.get_boundary::<T>(IteratorMode::Start)
1121    }
1122
1123    /// Gets the last (largest key) entry from the specified table.
1124    #[inline]
1125    pub fn last<T: Table>(&self) -> ProviderResult<Option<(T::Key, T::Value)>> {
1126        self.get_boundary::<T>(IteratorMode::End)
1127    }
1128
1129    /// Creates an iterator over all entries in the specified table.
1130    ///
1131    /// Returns decoded `(Key, Value)` pairs in key order.
1132    pub fn iter<T: Table>(&self) -> ProviderResult<RocksDBIter<'_, T>> {
1133        let cf = self.get_cf_handle::<T>()?;
1134        let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1135        Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1136    }
1137
1138    /// Creates an iterator starting from the given key (inclusive, seek forward).
1139    ///
1140    /// Returns decoded `(Key, Value)` pairs starting from the first key >= `key`.
1141    pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksDBIter<'_, T>> {
1142        let cf = self.get_cf_handle::<T>()?;
1143        let encoded_key = key.encode();
1144        let iter = self
1145            .0
1146            .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
1147        Ok(RocksDBIter { inner: iter, _marker: std::marker::PhantomData })
1148    }
1149
1150    /// Returns statistics for all column families in the database.
1151    ///
1152    /// Returns a vector of (`table_name`, `estimated_keys`, `estimated_size_bytes`) tuples.
1153    pub fn table_stats(&self) -> Vec<RocksDBTableStats> {
1154        self.0.table_stats()
1155    }
1156
1157    /// Returns the total size of WAL (Write-Ahead Log) files in bytes.
1158    ///
1159    /// This scans the `RocksDB` directory for `.log` files and sums their sizes.
1160    /// WAL files can be significant (e.g., 2.7GB observed) and are not included
1161    /// in `table_size`, `sst_size`, or `memtable_size` metrics.
1162    pub fn wal_size_bytes(&self) -> u64 {
1163        self.0.wal_size_bytes()
1164    }
1165
1166    /// Returns database-level statistics including per-table stats and WAL size.
1167    ///
1168    /// This combines [`Self::table_stats`] and [`Self::wal_size_bytes`] into a single struct.
1169    pub fn db_stats(&self) -> RocksDBStats {
1170        self.0.db_stats()
1171    }
1172
1173    /// Flushes pending writes for the specified tables to disk.
1174    ///
1175    /// This performs a flush of:
1176    /// 1. The column family memtables for the specified table names to SST files
1177    /// 2. The Write-Ahead Log (WAL) with sync
1178    ///
1179    /// After this call completes, all data for the specified tables is durably persisted to disk.
1180    ///
1181    /// # Panics
1182    /// Panics if the provider is in read-only mode.
1183    #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(tables = ?tables))]
1184    pub fn flush(&self, tables: &[&'static str]) -> ProviderResult<()> {
1185        let db = self.0.db_rw();
1186
1187        for cf_name in tables {
1188            if let Some(cf) = db.cf_handle(cf_name) {
1189                db.flush_cf(&cf).map_err(|e| {
1190                    ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1191                        info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1192                        operation: DatabaseWriteOperation::Flush,
1193                        table_name: cf_name,
1194                        key: Vec::new(),
1195                    })))
1196                })?;
1197            }
1198        }
1199
1200        db.flush_wal(true).map_err(|e| {
1201            ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
1202                info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
1203                operation: DatabaseWriteOperation::Flush,
1204                table_name: "WAL",
1205                key: Vec::new(),
1206            })))
1207        })?;
1208
1209        Ok(())
1210    }
1211
1212    /// Flushes and compacts all tables in `RocksDB`.
1213    ///
1214    /// This:
1215    /// 1. Flushes all column family memtables to SST files
1216    /// 2. Flushes the Write-Ahead Log (WAL) with sync
1217    /// 3. Triggers manual compaction on all column families to reclaim disk space
1218    ///
1219    /// Use this after large delete operations (like pruning) to reclaim disk space.
1220    ///
1221    /// # Panics
1222    /// Panics if the provider is in read-only mode.
1223    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1224    pub fn flush_and_compact(&self) -> ProviderResult<()> {
1225        self.flush(ROCKSDB_TABLES)?;
1226
1227        let db = self.0.db_rw();
1228
1229        for cf_name in ROCKSDB_TABLES {
1230            if let Some(cf) = db.cf_handle(cf_name) {
1231                db.compact_range_cf(&cf, None::<&[u8]>, None::<&[u8]>);
1232            }
1233        }
1234
1235        Ok(())
1236    }
1237
1238    /// Creates a raw iterator over all entries in the specified table.
1239    ///
1240    /// Returns raw `(key_bytes, value_bytes)` pairs without decoding.
1241    pub fn raw_iter<T: Table>(&self) -> ProviderResult<RocksDBRawIter<'_>> {
1242        let cf = self.get_cf_handle::<T>()?;
1243        let iter = self.0.iterator_cf(cf, IteratorMode::Start);
1244        Ok(RocksDBRawIter { inner: iter })
1245    }
1246
1247    /// Creates a raw key iterator positioned at `key`.
1248    pub(crate) fn raw_key_iter_from<T: Table>(
1249        &self,
1250        key: T::Key,
1251    ) -> ProviderResult<RocksDBRawKeyIter<'_>> {
1252        let cf = self.get_cf_handle::<T>()?;
1253        let encoded_key = key.encode();
1254        let mut iter = self.0.raw_iterator_cf(cf);
1255        iter.seek(encoded_key.as_ref());
1256        Ok(RocksDBRawKeyIter { inner: iter })
1257    }
1258
1259    /// Returns all account history shards for the given address in ascending key order.
1260    ///
1261    /// This is used for unwind operations where we need to scan all shards for an address
1262    /// and potentially delete or truncate them.
1263    pub fn account_history_shards(
1264        &self,
1265        address: Address,
1266    ) -> ProviderResult<Vec<(ShardedKey<Address>, BlockNumberList)>> {
1267        // Get the column family handle for the AccountsHistory table.
1268        let cf = self.get_cf_handle::<tables::AccountsHistory>()?;
1269
1270        // Build a seek key starting at the first shard (highest_block_number = 0) for this address.
1271        // ShardedKey is (address, highest_block_number) so this positions us at the beginning.
1272        let start_key = ShardedKey::new(address, 0u64);
1273        let start_bytes = start_key.encode();
1274        // The sentinel shard is the last possible key for this address.
1275        let last_bytes = ShardedKey::new(address, u64::MAX).encode();
1276
1277        // Create a forward iterator over this address's shards only, so an address without live
1278        // shards does not scan the tombstones of the addresses after it.
1279        let iter =
1280            self.0.iterator_cf_inclusive_range(cf, start_bytes.as_ref(), last_bytes.as_ref());
1281
1282        let mut result = Vec::new();
1283        for item in iter {
1284            match item {
1285                Ok((key_bytes, value_bytes)) => {
1286                    // Decode the sharded key to check if we're still on the same address.
1287                    let key = ShardedKey::<Address>::decode(&key_bytes)
1288                        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1289
1290                    // Stop when we reach a different address (keys are sorted by address first).
1291                    if key.key != address {
1292                        break;
1293                    }
1294
1295                    // Decompress the block number list stored in this shard.
1296                    let value = BlockNumberList::decompress(&value_bytes)
1297                        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1298
1299                    result.push((key, value));
1300                }
1301                Err(e) => {
1302                    return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1303                        message: e.to_string().into(),
1304                        code: -1,
1305                    })));
1306                }
1307            }
1308        }
1309
1310        Ok(result)
1311    }
1312
1313    /// Returns all storage history shards for the given `(address, storage_key)` pair.
1314    ///
1315    /// Iterates through all shards in ascending `highest_block_number` order until
1316    /// a different `(address, storage_key)` is encountered.
1317    pub fn storage_history_shards(
1318        &self,
1319        address: Address,
1320        storage_key: B256,
1321    ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1322        let cf = self.get_cf_handle::<tables::StoragesHistory>()?;
1323
1324        let start_key = StorageShardedKey::new(address, storage_key, 0u64);
1325        let start_bytes = start_key.encode();
1326        let last_bytes = StorageShardedKey::new(address, storage_key, u64::MAX).encode();
1327
1328        // Bounded to this slot's shards, see `account_history_shards`.
1329        let iter =
1330            self.0.iterator_cf_inclusive_range(cf, start_bytes.as_ref(), last_bytes.as_ref());
1331
1332        let mut result = Vec::new();
1333        for item in iter {
1334            match item {
1335                Ok((key_bytes, value_bytes)) => {
1336                    let key = StorageShardedKey::decode(&key_bytes)
1337                        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1338
1339                    if key.address != address || key.sharded_key.key != storage_key {
1340                        break;
1341                    }
1342
1343                    let value = BlockNumberList::decompress(&value_bytes)
1344                        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
1345
1346                    result.push((key, value));
1347                }
1348                Err(e) => {
1349                    return Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1350                        message: e.to_string().into(),
1351                        code: -1,
1352                    })));
1353                }
1354            }
1355        }
1356
1357        Ok(result)
1358    }
1359
1360    /// Unwinds account history indices for the given `(address, block_number)` pairs.
1361    ///
1362    /// Groups addresses by their minimum block number and calls the appropriate unwind
1363    /// operations. For each address, keeps only blocks less than the minimum block
1364    /// (i.e., removes the minimum block and all higher blocks).
1365    ///
1366    /// Returns a `WriteBatchWithTransaction` that can be committed later.
1367    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1368    pub fn unwind_account_history_indices(
1369        &self,
1370        last_indices: &[(Address, BlockNumber)],
1371    ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1372        let mut address_min_block: AddressMap<BlockNumber> =
1373            AddressMap::with_capacity_and_hasher(last_indices.len(), Default::default());
1374        for &(address, block_number) in last_indices {
1375            address_min_block
1376                .entry(address)
1377                .and_modify(|min| *min = (*min).min(block_number))
1378                .or_insert(block_number);
1379        }
1380
1381        let mut batch = self.batch();
1382        for (address, min_block) in address_min_block {
1383            match min_block.checked_sub(1) {
1384                Some(keep_to) => batch.unwind_account_history_to(address, keep_to)?,
1385                None => batch.clear_account_history(address)?,
1386            }
1387        }
1388
1389        Ok(batch.into_inner())
1390    }
1391
1392    /// Unwinds storage history indices for the given `(address, storage_key, block_number)` tuples.
1393    ///
1394    /// Groups by `(address, storage_key)` and finds the minimum block number for each.
1395    /// For each key, keeps only blocks less than the minimum block
1396    /// (i.e., removes the minimum block and all higher blocks).
1397    ///
1398    /// Returns a `WriteBatchWithTransaction` that can be committed later.
1399    pub fn unwind_storage_history_indices(
1400        &self,
1401        storage_changesets: &[(Address, B256, BlockNumber)],
1402    ) -> ProviderResult<WriteBatchWithTransaction<true>> {
1403        let mut key_min_block: HashMap<(Address, B256), BlockNumber> =
1404            HashMap::with_capacity_and_hasher(storage_changesets.len(), Default::default());
1405        for &(address, storage_key, block_number) in storage_changesets {
1406            key_min_block
1407                .entry((address, storage_key))
1408                .and_modify(|min| *min = (*min).min(block_number))
1409                .or_insert(block_number);
1410        }
1411
1412        let mut batch = self.batch();
1413        for ((address, storage_key), min_block) in key_min_block {
1414            match min_block.checked_sub(1) {
1415                Some(keep_to) => batch.unwind_storage_history_to(address, storage_key, keep_to)?,
1416                None => batch.clear_storage_history(address, storage_key)?,
1417            }
1418        }
1419
1420        Ok(batch.into_inner())
1421    }
1422
1423    /// Writes a batch of operations atomically.
1424    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1425    pub fn write_batch<F>(&self, f: F) -> ProviderResult<()>
1426    where
1427        F: FnOnce(&mut RocksDBBatch<'_>) -> ProviderResult<()>,
1428    {
1429        self.execute_with_operation_metric(RocksDBOperation::BatchWrite, "Batch", |this| {
1430            let mut batch_handle = this.batch();
1431            f(&mut batch_handle)?;
1432            batch_handle.commit()
1433        })
1434    }
1435
1436    /// Commits a raw `WriteBatchWithTransaction` to `RocksDB`.
1437    ///
1438    /// This is used when the batch was extracted via [`RocksDBBatch::into_inner`]
1439    /// and needs to be committed at a later point (e.g., at provider commit time).
1440    ///
1441    /// # Panics
1442    /// Panics if the provider is in read-only mode.
1443    #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = batch.len(), batch_size = batch.size_in_bytes()))]
1444    pub fn commit_batch(&self, batch: WriteBatchWithTransaction<true>) -> ProviderResult<()> {
1445        self.0.db_rw().write_opt(batch, &synced_write_options()).map_err(|e| {
1446            ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
1447                message: e.to_string().into(),
1448                code: -1,
1449            }))
1450        })
1451    }
1452
1453    /// Writes all `RocksDB` data for multiple blocks in parallel.
1454    ///
1455    /// This handles transaction hash numbers, account history, and storage history based on
1456    /// the provided storage settings. Each operation runs in parallel with its own batch,
1457    /// pushing to `ctx.pending_batches` for later commit.
1458    #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(num_blocks = blocks.len(), first_block = ctx.first_block_number))]
1459    pub(crate) fn write_blocks_data<N: reth_node_types::NodePrimitives>(
1460        &self,
1461        blocks: &[ExecutedBlock<N>],
1462        tx_nums: &[TxNumber],
1463        ctx: RocksDBWriteCtx,
1464        runtime: &reth_tasks::Runtime,
1465    ) -> ProviderResult<()> {
1466        if !ctx.storage_settings.storage_v2 {
1467            return Ok(());
1468        }
1469
1470        let mut r_tx_hash = None;
1471        let mut r_account_history = None;
1472        let mut r_storage_history = None;
1473
1474        let write_tx_hash =
1475            ctx.storage_settings.storage_v2 && ctx.prune_tx_lookup.is_none_or(|m| !m.is_full());
1476        let write_account_history = ctx.storage_settings.storage_v2;
1477        let write_storage_history = ctx.storage_settings.storage_v2;
1478
1479        // Propagate tracing context into rayon-spawned threads so that RocksDB
1480        // write spans appear as children of write_blocks_data in traces.
1481        let span = tracing::Span::current();
1482        runtime.storage_pool().in_place_scope(|s| {
1483            if write_tx_hash {
1484                s.spawn(|_| {
1485                    let _guard = span.enter();
1486                    r_tx_hash = Some(self.write_tx_hash_numbers(blocks, tx_nums, &ctx));
1487                });
1488            }
1489
1490            if write_account_history {
1491                s.spawn(|_| {
1492                    let _guard = span.enter();
1493                    r_account_history = Some(self.write_account_history(blocks, &ctx));
1494                });
1495            }
1496
1497            if write_storage_history {
1498                s.spawn(|_| {
1499                    let _guard = span.enter();
1500                    r_storage_history = Some(self.write_storage_history(blocks, &ctx));
1501                });
1502            }
1503        });
1504
1505        if write_tx_hash {
1506            r_tx_hash.ok_or_else(|| {
1507                ProviderError::Database(DatabaseError::Other(
1508                    "rocksdb tx-hash write thread panicked".into(),
1509                ))
1510            })??;
1511        }
1512        if write_account_history {
1513            r_account_history.ok_or_else(|| {
1514                ProviderError::Database(DatabaseError::Other(
1515                    "rocksdb account-history write thread panicked".into(),
1516                ))
1517            })??;
1518        }
1519        if write_storage_history {
1520            r_storage_history.ok_or_else(|| {
1521                ProviderError::Database(DatabaseError::Other(
1522                    "rocksdb storage-history write thread panicked".into(),
1523                ))
1524            })??;
1525        }
1526
1527        Ok(())
1528    }
1529
1530    /// Writes transaction hash to number mappings for the given blocks.
1531    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1532    fn write_tx_hash_numbers<N: reth_node_types::NodePrimitives>(
1533        &self,
1534        blocks: &[ExecutedBlock<N>],
1535        tx_nums: &[TxNumber],
1536        ctx: &RocksDBWriteCtx,
1537    ) -> ProviderResult<()> {
1538        let mut batch = self.batch();
1539        for (block, &first_tx_num) in blocks.iter().zip(tx_nums) {
1540            let body = block.recovered_block().body();
1541            for (tx_num, transaction) in (first_tx_num..).zip(body.transactions_iter()) {
1542                batch.put::<tables::TransactionHashNumbers>(*transaction.tx_hash(), &tx_num)?;
1543            }
1544        }
1545        ctx.pending_batches.lock().push(batch.into_inner());
1546        Ok(())
1547    }
1548
1549    /// Writes account history indices for the given blocks.
1550    ///
1551    /// Derives history indices from reverts (same source as changesets) to ensure consistency.
1552    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1553    fn write_account_history<N: reth_node_types::NodePrimitives>(
1554        &self,
1555        blocks: &[ExecutedBlock<N>],
1556        ctx: &RocksDBWriteCtx,
1557    ) -> ProviderResult<()> {
1558        let mut batch = self.batch();
1559        let mut account_history: BTreeMap<Address, Vec<u64>> = BTreeMap::new();
1560
1561        for (block_idx, block) in blocks.iter().enumerate() {
1562            let block_number = ctx.first_block_number + block_idx as u64;
1563            let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1564
1565            // Iterate through account reverts - these are exactly the accounts that have
1566            // changesets written, ensuring history indices match changeset entries.
1567            for account_block_reverts in reverts.accounts {
1568                for (address, _) in account_block_reverts {
1569                    account_history.entry(address).or_default().push(block_number);
1570                }
1571            }
1572        }
1573
1574        // Write account history using proper shard append logic
1575        for (address, indices) in account_history {
1576            batch.append_account_history_shard(address, indices)?;
1577        }
1578        ctx.pending_batches.lock().push(batch.into_inner());
1579        Ok(())
1580    }
1581
1582    /// Writes storage history indices for the given blocks.
1583    ///
1584    /// Derives history indices from reverts (same source as changesets) to ensure consistency.
1585    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
1586    fn write_storage_history<N: reth_node_types::NodePrimitives>(
1587        &self,
1588        blocks: &[ExecutedBlock<N>],
1589        ctx: &RocksDBWriteCtx,
1590    ) -> ProviderResult<()> {
1591        let mut storage_history: BTreeMap<(Address, B256), Vec<u64>> = BTreeMap::new();
1592
1593        for (block_idx, block) in blocks.iter().enumerate() {
1594            let block_number = ctx.first_block_number + block_idx as u64;
1595            let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
1596
1597            // Iterate through storage reverts - these are exactly the slots that have
1598            // changesets written, ensuring history indices match changeset entries.
1599            for storage_block_reverts in reverts.storage {
1600                for revert in storage_block_reverts {
1601                    for (slot, _) in revert.storage_revert {
1602                        let plain_key = B256::from(slot);
1603                        storage_history
1604                            .entry((revert.address, plain_key))
1605                            .or_default()
1606                            .push(block_number);
1607                    }
1608                }
1609            }
1610        }
1611
1612        let shard_puts = storage_history
1613            .into_par_iter()
1614            .map(|((address, slot), indices)| {
1615                self.storage_history_shards_to_put(address, slot, indices)
1616            })
1617            .collect::<ProviderResult<Vec<_>>>()?;
1618
1619        let mut batch = self.batch();
1620        for shards in shard_puts {
1621            for (key, shard) in shards {
1622                batch.put::<tables::StoragesHistory>(key, &shard)?;
1623            }
1624        }
1625        ctx.pending_batches.lock().push(batch.into_inner());
1626        Ok(())
1627    }
1628
1629    /// Prepares storage history shard writes by reading the current last shard and appending
1630    /// indices.
1631    fn storage_history_shards_to_put(
1632        &self,
1633        address: Address,
1634        storage_key: B256,
1635        indices: Vec<u64>,
1636    ) -> ProviderResult<Vec<(StorageShardedKey, BlockNumberList)>> {
1637        if indices.is_empty() {
1638            return Ok(Vec::new());
1639        }
1640
1641        debug_assert!(
1642            indices.windows(2).all(|w| w[0] < w[1]),
1643            "indices must be strictly increasing: {:?}",
1644            indices
1645        );
1646
1647        let last_key = StorageShardedKey::last(address, storage_key);
1648        let last_shard_opt = self.get::<tables::StoragesHistory>(last_key.clone())?;
1649        let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
1650
1651        last_shard.append(indices).map_err(ProviderError::other)?;
1652
1653        if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
1654            return Ok(vec![(last_key, last_shard)]);
1655        }
1656
1657        let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
1658        let mut chunks_peekable = chunks.into_iter().peekable();
1659        let mut shards = Vec::new();
1660
1661        while let Some(chunk) = chunks_peekable.next() {
1662            let shard = BlockNumberList::new_pre_sorted(chunk);
1663            let highest_block_number = if chunks_peekable.peek().is_some() {
1664                shard.iter().next_back().expect("`chunks` does not return empty list")
1665            } else {
1666                u64::MAX
1667            };
1668
1669            shards
1670                .push((StorageShardedKey::new(address, storage_key, highest_block_number), shard));
1671        }
1672
1673        Ok(shards)
1674    }
1675}
1676
1677/// A point-in-time read snapshot of the `RocksDB` database.
1678///
1679/// All reads through this snapshot see a consistent view of the database at the point
1680/// the snapshot was created, regardless of concurrent writes. This is the primary reader
1681/// used by [`EitherReader::RocksDB`](crate::either_writer::EitherReader) for history lookups.
1682///
1683/// Lighter weight than [`RocksTx`] — no transaction overhead, no write support.
1684pub struct RocksReadSnapshot<'db> {
1685    /// Raw iterator reused across account history lookups.
1686    ///
1687    /// Declared before `inner` so it is dropped before the snapshot it reads from.
1688    accounts_history_iter: Mutex<Option<RocksDBRawIterEnum<'db>>>,
1689    /// Raw iterator reused across storage history lookups.
1690    ///
1691    /// Declared before `inner` so it is dropped before the snapshot it reads from.
1692    storages_history_iter: Mutex<Option<RocksDBRawIterEnum<'db>>>,
1693    inner: RocksReadSnapshotInner<'db>,
1694    provider: &'db RocksDBProviderInner,
1695}
1696
1697/// Inner enum to hold the snapshot for either read-write or secondary mode.
1698enum RocksReadSnapshotInner<'db> {
1699    /// Snapshot from read-write `OptimisticTransactionDB`.
1700    ReadWrite(SnapshotWithThreadMode<'db, OptimisticTransactionDB>),
1701    /// Direct reads from a secondary `DB` instance (no snapshot).
1702    Secondary(&'db DB),
1703}
1704
1705impl fmt::Debug for RocksReadSnapshot<'_> {
1706    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1707        f.debug_struct("RocksReadSnapshot")
1708            .field("provider", &self.provider)
1709            .finish_non_exhaustive()
1710    }
1711}
1712
1713impl<'db> RocksReadSnapshot<'db> {
1714    /// Gets the column family handle for a table.
1715    fn cf_handle<T: Table>(&self) -> Result<&'db rocksdb::ColumnFamily, DatabaseError> {
1716        self.provider.cf_handle::<T>()
1717    }
1718
1719    /// Creates a raw iterator over `cf` that observes this snapshot.
1720    ///
1721    /// The iterator is created from the database handle rather than from the snapshot value, so
1722    /// its lifetime is tied to the database and it can be cached in `self`. The snapshot is
1723    /// attached through [`ReadOptions`], giving the same point-in-time view as reading through the
1724    /// snapshot directly.
1725    fn new_raw_iterator_cf(&self, cf: &rocksdb::ColumnFamily) -> RocksDBRawIterEnum<'db> {
1726        match &self.inner {
1727            RocksReadSnapshotInner::ReadWrite(snap) => {
1728                let mut readopts = ReadOptions::default();
1729                readopts.set_snapshot(snap);
1730                RocksDBRawIterEnum::ReadWrite(
1731                    self.provider.db_rw().raw_iterator_cf_opt(cf, readopts),
1732                )
1733            }
1734            RocksReadSnapshotInner::Secondary(db) => {
1735                RocksDBRawIterEnum::ReadOnly((*db).raw_iterator_cf(cf))
1736            }
1737        }
1738    }
1739
1740    /// Gets a value from the specified table.
1741    pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
1742        let encoded_key = key.encode();
1743        let cf = self.cf_handle::<T>()?;
1744        let result = match &self.inner {
1745            RocksReadSnapshotInner::ReadWrite(snap) => snap.get_cf(cf, encoded_key.as_ref()),
1746            RocksReadSnapshotInner::Secondary(db) => db.get_cf(cf, encoded_key.as_ref()),
1747        }
1748        .map_err(|e| {
1749            ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1750                message: e.to_string().into(),
1751                code: -1,
1752            }))
1753        })?;
1754
1755        Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
1756    }
1757
1758    /// Lookup account history and return [`HistoryInfo`] directly.
1759    ///
1760    /// `visible_tip` is the highest block considered visible from the companion MDBX snapshot.
1761    /// History entries above it are ignored even if they already exist in `RocksDB`.
1762    pub fn account_history_info(
1763        &self,
1764        address: Address,
1765        block_number: BlockNumber,
1766        lowest_available_block_number: Option<BlockNumber>,
1767        visible_tip: BlockNumber,
1768    ) -> ProviderResult<HistoryInfo> {
1769        let key = ShardedKey::new(address, block_number);
1770        self.history_info::<tables::AccountsHistory>(
1771            &self.accounts_history_iter,
1772            key.encode().as_ref(),
1773            block_number,
1774            lowest_available_block_number,
1775            visible_tip,
1776            |key_bytes| Ok(<ShardedKey<Address> as Decode>::decode(key_bytes)?.key == address),
1777            |prev_bytes| {
1778                <ShardedKey<Address> as Decode>::decode(prev_bytes)
1779                    .map(|k| k.key == address)
1780                    .unwrap_or(false)
1781            },
1782        )
1783    }
1784
1785    /// Lookup storage history and return [`HistoryInfo`] directly.
1786    ///
1787    /// `visible_tip` is the highest block considered visible from the companion MDBX snapshot.
1788    /// History entries above it are ignored even if they already exist in `RocksDB`.
1789    pub fn storage_history_info(
1790        &self,
1791        address: Address,
1792        storage_key: B256,
1793        block_number: BlockNumber,
1794        lowest_available_block_number: Option<BlockNumber>,
1795        visible_tip: BlockNumber,
1796    ) -> ProviderResult<HistoryInfo> {
1797        let key = StorageShardedKey::new(address, storage_key, block_number);
1798        self.history_info::<tables::StoragesHistory>(
1799            &self.storages_history_iter,
1800            key.encode().as_ref(),
1801            block_number,
1802            lowest_available_block_number,
1803            visible_tip,
1804            |key_bytes| {
1805                let k = <StorageShardedKey as Decode>::decode(key_bytes)?;
1806                Ok(k.address == address && k.sharded_key.key == storage_key)
1807            },
1808            |prev_bytes| {
1809                <StorageShardedKey as Decode>::decode(prev_bytes)
1810                    .map(|k| k.address == address && k.sharded_key.key == storage_key)
1811                    .unwrap_or(false)
1812            },
1813        )
1814    }
1815
1816    /// Generic history lookup using the snapshot's raw iterator.
1817    ///
1818    /// The result is derived from the history that is visible through `visible_tip`, not from the
1819    /// full contents of `RocksDB`. This lets a reader combine an older MDBX snapshot with a newer
1820    /// Rocks snapshot without routing through history entries that MDBX cannot see yet.
1821    ///
1822    /// `iter_cache` holds the raw iterator for `T`'s column family. Seeking an existing iterator
1823    /// is much cheaper than constructing one, so the iterator is created on the first lookup and
1824    /// reused by every later lookup through this snapshot. A lookup that finds the cache held by
1825    /// another thread uses a private iterator instead of waiting.
1826    #[expect(clippy::too_many_arguments)]
1827    fn history_info<T>(
1828        &self,
1829        iter_cache: &Mutex<Option<RocksDBRawIterEnum<'db>>>,
1830        encoded_key: &[u8],
1831        block_number: BlockNumber,
1832        lowest_available_block_number: Option<BlockNumber>,
1833        visible_tip: BlockNumber,
1834        key_matches: impl FnOnce(&[u8]) -> Result<bool, reth_db_api::DatabaseError>,
1835        prev_key_matches: impl Fn(&[u8]) -> bool,
1836    ) -> ProviderResult<HistoryInfo>
1837    where
1838        T: Table<Value = BlockNumberList>,
1839    {
1840        let is_maybe_pruned = lowest_available_block_number.is_some();
1841        let fallback = || {
1842            Ok(if is_maybe_pruned {
1843                HistoryInfo::MaybeInPlainState
1844            } else {
1845                HistoryInfo::NotYetWritten
1846            })
1847        };
1848
1849        let cf = self.cf_handle::<T>()?;
1850        // A lookup on another thread may be holding the cached iterator; a private iterator costs
1851        // what every lookup used to cost, whereas waiting would serialize the two lookups.
1852        let mut guard = iter_cache.try_lock();
1853        let mut private_iter;
1854        let iter = match guard.as_mut() {
1855            Some(cached) => cached.get_or_insert_with(|| self.new_raw_iterator_cf(cf)),
1856            None => {
1857                private_iter = self.new_raw_iterator_cf(cf);
1858                &mut private_iter
1859            }
1860        };
1861
1862        iter.seek(encoded_key);
1863        iter.status().map_err(|e| {
1864            ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1865                message: e.to_string().into(),
1866                code: -1,
1867            }))
1868        })?;
1869
1870        if !iter.valid() {
1871            return fallback();
1872        }
1873
1874        let Some(key_bytes) = iter.key() else {
1875            return fallback();
1876        };
1877        if !key_matches(key_bytes)? {
1878            return fallback();
1879        }
1880
1881        let Some(value_bytes) = iter.value() else {
1882            return fallback();
1883        };
1884        let chunk = BlockNumberList::decompress(value_bytes)?;
1885
1886        let (rank, found_block) = compute_history_rank(&chunk, block_number);
1887        // Ignore later Rocks history that is ahead of the companion MDBX snapshot.
1888        let found_block = found_block.filter(|block| *block <= visible_tip);
1889
1890        let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) {
1891            iter.prev();
1892            iter.status().map_err(|e| {
1893                ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
1894                    message: e.to_string().into(),
1895                    code: -1,
1896                }))
1897            })?;
1898            let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches);
1899
1900            // If the current shard only contains history above `visible_tip`, there is no usable
1901            // later change. Without a previous shard for the same key, fall back to the existing
1902            // not-written / maybe-pruned result instead of routing into plain state.
1903            if found_block.is_none() && !has_prev {
1904                return fallback()
1905            }
1906
1907            !has_prev
1908        } else {
1909            false
1910        };
1911
1912        Ok(HistoryInfo::from_lookup(
1913            found_block,
1914            is_before_first_write,
1915            lowest_available_block_number,
1916        ))
1917    }
1918}
1919
1920/// A [`RocksReadSnapshot`] that owns the [`RocksDBProvider`] handle it reads through.
1921///
1922/// [`RocksDBProvider::snapshot`] borrows the provider, so a reader that wants to keep a single
1923/// snapshot alive has to keep the provider handle next to it. This type does that, which lets the
1924/// snapshot cache its history iterators across lookups.
1925pub struct OwnedRocksReadSnapshot {
1926    /// Borrows the allocation retained by `provider`. Declared first so the snapshot and its
1927    /// iterators are released before the owning database handle. Boxing keeps its internal
1928    /// references out of the movable owner, including when the owner is passed by value.
1929    snapshot: Box<RocksReadSnapshot<'static>>,
1930    provider: RocksDBProvider,
1931}
1932
1933impl fmt::Debug for OwnedRocksReadSnapshot {
1934    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1935        f.debug_struct("OwnedRocksReadSnapshot").field("provider", &self.provider).finish()
1936    }
1937}
1938
1939impl OwnedRocksReadSnapshot {
1940    fn new(provider: &RocksDBProvider) -> Self {
1941        let provider = provider.clone();
1942        let snapshot = provider.snapshot();
1943        // SAFETY: Every reference in `snapshot` points into the Arc allocation retained by
1944        // `provider`, not into the movable provider handle. The handle is private and never
1945        // replaced, and `snapshot` is dropped before it. `as_snapshot` restricts access to the
1946        // lifetime of a borrow of this owner.
1947        let snapshot = unsafe {
1948            std::mem::transmute::<RocksReadSnapshot<'_>, RocksReadSnapshot<'static>>(snapshot)
1949        };
1950        Self { snapshot: Box::new(snapshot), provider }
1951    }
1952
1953    /// Returns the borrowed snapshot.
1954    pub fn as_snapshot(&self) -> &RocksReadSnapshot<'_> {
1955        // SAFETY: shortening the snapshot's lifetime to this borrow of `self` is sound; the owned
1956        // provider handle keeps the database open for at least that long.
1957        unsafe {
1958            std::mem::transmute::<&RocksReadSnapshot<'static>, &RocksReadSnapshot<'_>>(
1959                &self.snapshot,
1960            )
1961        }
1962    }
1963}
1964
1965/// Outcome of pruning a history shard in `RocksDB`.
1966#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1967pub enum PruneShardOutcome {
1968    /// Shard was deleted entirely.
1969    Deleted,
1970    /// Shard was updated with filtered block numbers.
1971    Updated,
1972    /// Shard was unchanged (no blocks <= `to_block`).
1973    Unchanged,
1974}
1975
1976/// Tracks pruning outcomes for batch operations.
1977#[derive(Debug, Default, Clone, Copy)]
1978pub struct PrunedIndices {
1979    /// Number of shards completely deleted.
1980    pub deleted: usize,
1981    /// Number of shards that were updated (filtered but still have entries).
1982    pub updated: usize,
1983    /// Number of shards that were unchanged.
1984    pub unchanged: usize,
1985}
1986
1987/// Handle for building a batch of operations atomically.
1988///
1989/// Uses `WriteBatchWithTransaction` for atomic writes without full transaction overhead.
1990/// Unlike [`RocksTx`], this does NOT support read-your-writes. Use for write-only flows
1991/// where you don't need to read back uncommitted data within the same operation
1992/// (e.g., history index writes).
1993///
1994/// When `auto_commit_threshold` is set, the batch will automatically commit and reset
1995/// when the batch size exceeds the threshold. This prevents OOM during large bulk writes.
1996#[must_use = "batch must be committed"]
1997pub struct RocksDBBatch<'a> {
1998    provider: &'a RocksDBProvider,
1999    inner: WriteBatchWithTransaction<true>,
2000    buf: Vec<u8>,
2001    /// If set, batch auto-commits when size exceeds this threshold (in bytes).
2002    auto_commit_threshold: Option<usize>,
2003}
2004
2005impl fmt::Debug for RocksDBBatch<'_> {
2006    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2007        f.debug_struct("RocksDBBatch")
2008            .field("provider", &self.provider)
2009            .field("batch", &"<WriteBatchWithTransaction>")
2010            // Number of operations in this batch
2011            .field("length", &self.inner.len())
2012            // Total serialized size (encoded key + compressed value + metadata) of this batch
2013            // in bytes
2014            .field("size_in_bytes", &self.inner.size_in_bytes())
2015            .finish()
2016    }
2017}
2018
2019impl<'a> RocksDBBatch<'a> {
2020    /// Puts a value into the batch.
2021    ///
2022    /// If auto-commit is enabled and the batch exceeds the threshold, commits and resets.
2023    pub fn put<T: Table>(&mut self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
2024        let encoded_key = key.encode();
2025        self.put_encoded::<T>(&encoded_key, value)
2026    }
2027
2028    /// Puts a value into the batch using pre-encoded key.
2029    ///
2030    /// If auto-commit is enabled and the batch exceeds the threshold, commits and resets.
2031    pub fn put_encoded<T: Table>(
2032        &mut self,
2033        key: &<T::Key as Encode>::Encoded,
2034        value: &T::Value,
2035    ) -> ProviderResult<()> {
2036        let value_bytes = compress_to_buf_or_ref!(self.buf, value).unwrap_or(&self.buf);
2037        self.inner.put_cf(self.provider.get_cf_handle::<T>()?, key, value_bytes);
2038        self.maybe_auto_commit()?;
2039        Ok(())
2040    }
2041
2042    /// Deletes a value from the batch.
2043    ///
2044    /// If auto-commit is enabled and the batch exceeds the threshold, commits and resets.
2045    pub fn delete<T: Table>(&mut self, key: T::Key) -> ProviderResult<()> {
2046        self.inner.delete_cf(self.provider.get_cf_handle::<T>()?, key.encode().as_ref());
2047        self.maybe_auto_commit()?;
2048        Ok(())
2049    }
2050
2051    /// Commits and resets the batch if it exceeds the auto-commit threshold.
2052    ///
2053    /// This is called after each `put` or `delete` operation to prevent unbounded memory growth.
2054    /// Returns immediately if auto-commit is disabled or threshold not reached.
2055    fn maybe_auto_commit(&mut self) -> ProviderResult<()> {
2056        if let Some(threshold) = self.auto_commit_threshold &&
2057            self.inner.size_in_bytes() >= threshold
2058        {
2059            tracing::debug!(
2060                target: "providers::rocksdb",
2061                batch_size = self.inner.size_in_bytes(),
2062                threshold,
2063                "Auto-committing RocksDB batch"
2064            );
2065            let old_batch = std::mem::take(&mut self.inner);
2066            self.provider.0.db_rw().write_opt(old_batch, &synced_write_options()).map_err(|e| {
2067                ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2068                    message: e.to_string().into(),
2069                    code: -1,
2070                }))
2071            })?;
2072        }
2073        Ok(())
2074    }
2075
2076    /// Commits the batch to the database.
2077    ///
2078    /// This consumes the batch and writes all operations atomically to `RocksDB`.
2079    ///
2080    /// # Panics
2081    /// Panics if the provider is in read-only mode.
2082    #[instrument(level = "debug", target = "providers::rocksdb", skip_all, fields(batch_len = self.inner.len(), batch_size = self.inner.size_in_bytes()))]
2083    pub fn commit(self) -> ProviderResult<()> {
2084        self.provider.0.db_rw().write_opt(self.inner, &synced_write_options()).map_err(|e| {
2085            ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2086                message: e.to_string().into(),
2087                code: -1,
2088            }))
2089        })
2090    }
2091
2092    /// Returns the number of write operations (puts + deletes) queued in this batch.
2093    pub fn len(&self) -> usize {
2094        self.inner.len()
2095    }
2096
2097    /// Returns `true` if the batch contains no operations.
2098    pub fn is_empty(&self) -> bool {
2099        self.inner.is_empty()
2100    }
2101
2102    /// Returns the size of the batch in bytes.
2103    pub fn size_in_bytes(&self) -> usize {
2104        self.inner.size_in_bytes()
2105    }
2106
2107    /// Returns a reference to the underlying `RocksDB` provider.
2108    pub const fn provider(&self) -> &RocksDBProvider {
2109        self.provider
2110    }
2111
2112    /// Consumes the batch and returns the underlying `WriteBatchWithTransaction`.
2113    ///
2114    /// This is used to defer commits to the provider level.
2115    pub fn into_inner(self) -> WriteBatchWithTransaction<true> {
2116        self.inner
2117    }
2118
2119    /// Gets a value from the database.
2120    ///
2121    /// **Important constraint:** This reads only committed state, not pending writes in this
2122    /// batch or other pending batches in `pending_rocksdb_batches`.
2123    pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
2124        self.provider.get::<T>(key)
2125    }
2126
2127    /// Appends indices to an account history shard with proper shard management.
2128    ///
2129    /// Loads the existing shard (if any), appends new indices, and rechunks into
2130    /// multiple shards if needed (respecting `NUM_OF_INDICES_IN_SHARD` limit).
2131    ///
2132    /// # Requirements
2133    ///
2134    /// - The `indices` MUST be strictly increasing and contain no duplicates.
2135    /// - This method MUST only be called once per address per batch. The batch reads existing
2136    ///   shards from committed DB state, not from pending writes. Calling twice for the same
2137    ///   address will cause the second call to overwrite the first.
2138    pub fn append_account_history_shard(
2139        &mut self,
2140        address: Address,
2141        indices: impl IntoIterator<Item = u64>,
2142    ) -> ProviderResult<()> {
2143        let indices: Vec<u64> = indices.into_iter().collect();
2144
2145        if indices.is_empty() {
2146            return Ok(());
2147        }
2148
2149        debug_assert!(
2150            indices.windows(2).all(|w| w[0] < w[1]),
2151            "indices must be strictly increasing: {:?}",
2152            indices
2153        );
2154
2155        let last_key = ShardedKey::new(address, u64::MAX);
2156        let last_shard_opt = self.provider.get::<tables::AccountsHistory>(last_key.clone())?;
2157        let mut last_shard = last_shard_opt.unwrap_or_else(BlockNumberList::empty);
2158
2159        last_shard.append(indices).map_err(ProviderError::other)?;
2160
2161        // Fast path: all indices fit in one shard
2162        if last_shard.len() <= NUM_OF_INDICES_IN_SHARD as u64 {
2163            self.put::<tables::AccountsHistory>(last_key, &last_shard)?;
2164            return Ok(());
2165        }
2166
2167        // Slow path: rechunk into multiple shards
2168        let chunks = last_shard.iter().chunks(NUM_OF_INDICES_IN_SHARD);
2169        let mut chunks_peekable = chunks.into_iter().peekable();
2170
2171        while let Some(chunk) = chunks_peekable.next() {
2172            let shard = BlockNumberList::new_pre_sorted(chunk);
2173            let highest_block_number = if chunks_peekable.peek().is_some() {
2174                shard.iter().next_back().expect("`chunks` does not return empty list")
2175            } else {
2176                u64::MAX
2177            };
2178
2179            self.put::<tables::AccountsHistory>(
2180                ShardedKey::new(address, highest_block_number),
2181                &shard,
2182            )?;
2183        }
2184
2185        Ok(())
2186    }
2187
2188    /// Appends indices to a storage history shard with proper shard management.
2189    ///
2190    /// Loads the existing shard (if any), appends new indices, and rechunks into
2191    /// multiple shards if needed (respecting `NUM_OF_INDICES_IN_SHARD` limit).
2192    ///
2193    /// # Requirements
2194    ///
2195    /// - The `indices` MUST be strictly increasing and contain no duplicates.
2196    /// - This method MUST only be called once per (address, `storage_key`) pair per batch. The
2197    ///   batch reads existing shards from committed DB state, not from pending writes. Calling
2198    ///   twice for the same key will cause the second call to overwrite the first.
2199    pub fn append_storage_history_shard(
2200        &mut self,
2201        address: Address,
2202        storage_key: B256,
2203        indices: impl IntoIterator<Item = u64>,
2204    ) -> ProviderResult<()> {
2205        let indices: Vec<u64> = indices.into_iter().collect();
2206
2207        for (key, shard) in
2208            self.provider.storage_history_shards_to_put(address, storage_key, indices)?
2209        {
2210            self.put::<tables::StoragesHistory>(key, &shard)?;
2211        }
2212
2213        Ok(())
2214    }
2215
2216    /// Unwinds account history for the given address, keeping only blocks <= `keep_to`.
2217    ///
2218    /// Mirrors MDBX `unwind_history_shards` behavior:
2219    /// - Deletes shards entirely above `keep_to`
2220    /// - Truncates boundary shards and re-keys to `u64::MAX` sentinel
2221    /// - Preserves shards entirely below `keep_to`
2222    pub fn unwind_account_history_to(
2223        &mut self,
2224        address: Address,
2225        keep_to: BlockNumber,
2226    ) -> ProviderResult<()> {
2227        let shards = self.provider.account_history_shards(address)?;
2228        if shards.is_empty() {
2229            return Ok(());
2230        }
2231
2232        // Find the first shard that might contain blocks > keep_to.
2233        // A shard is affected if it's the sentinel (u64::MAX) or its highest_block_number > keep_to
2234        let boundary_idx = shards.iter().position(|(key, _)| {
2235            key.highest_block_number == u64::MAX || key.highest_block_number > keep_to
2236        });
2237
2238        // Repair path: no shards affected means all blocks <= keep_to, just ensure sentinel exists
2239        let Some(boundary_idx) = boundary_idx else {
2240            let (last_key, last_value) = shards.last().expect("shards is non-empty");
2241            if last_key.highest_block_number != u64::MAX {
2242                self.delete::<tables::AccountsHistory>(last_key.clone())?;
2243                self.put::<tables::AccountsHistory>(
2244                    ShardedKey::new(address, u64::MAX),
2245                    last_value,
2246                )?;
2247            }
2248            return Ok(());
2249        };
2250
2251        // Delete all shards strictly after the boundary (they are entirely > keep_to)
2252        for (key, _) in shards.iter().skip(boundary_idx + 1) {
2253            self.delete::<tables::AccountsHistory>(key.clone())?;
2254        }
2255
2256        // Process the boundary shard: filter out blocks > keep_to
2257        let (boundary_key, boundary_list) = &shards[boundary_idx];
2258
2259        // Delete the boundary shard (we'll either drop it or rewrite at u64::MAX)
2260        self.delete::<tables::AccountsHistory>(boundary_key.clone())?;
2261
2262        // Build truncated list once; check emptiness directly (avoids double iteration)
2263        let new_last =
2264            BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2265
2266        if new_last.is_empty() {
2267            // Boundary shard is now empty. Previous shard becomes the last and must be keyed
2268            // u64::MAX.
2269            if boundary_idx == 0 {
2270                // Nothing left for this address
2271                return Ok(());
2272            }
2273
2274            let (prev_key, prev_value) = &shards[boundary_idx - 1];
2275            if prev_key.highest_block_number != u64::MAX {
2276                self.delete::<tables::AccountsHistory>(prev_key.clone())?;
2277                self.put::<tables::AccountsHistory>(
2278                    ShardedKey::new(address, u64::MAX),
2279                    prev_value,
2280                )?;
2281            }
2282            return Ok(());
2283        }
2284
2285        self.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &new_last)?;
2286
2287        Ok(())
2288    }
2289
2290    /// Prunes history shards, removing blocks <= `to_block`.
2291    ///
2292    /// Generic implementation for both account and storage history pruning.
2293    /// Mirrors MDBX `prune_shard` semantics. After pruning, the last remaining shard
2294    /// (if any) will have the sentinel key (`u64::MAX`).
2295    ///
2296    /// `shards_complete` must be `false` when `shards` stops short of the key's last shard, so the
2297    /// survivor is not re-keyed over a sentinel that is still on disk.
2298    #[expect(clippy::too_many_arguments)]
2299    fn prune_history_shards_inner<K>(
2300        &mut self,
2301        shards: Vec<(K, BlockNumberList)>,
2302        shards_complete: bool,
2303        to_block: BlockNumber,
2304        get_highest: impl Fn(&K) -> u64,
2305        is_sentinel: impl Fn(&K) -> bool,
2306        delete_shard: impl Fn(&mut Self, K) -> ProviderResult<()>,
2307        put_shard: impl Fn(&mut Self, K, &BlockNumberList) -> ProviderResult<()>,
2308        create_sentinel: impl Fn() -> K,
2309    ) -> ProviderResult<PruneShardOutcome>
2310    where
2311        K: Clone,
2312    {
2313        if shards.is_empty() {
2314            return Ok(PruneShardOutcome::Unchanged);
2315        }
2316
2317        let mut deleted = false;
2318        let mut updated = false;
2319        let mut last_remaining: Option<(K, BlockNumberList)> = None;
2320
2321        for (key, mut block_list) in shards {
2322            if !is_sentinel(&key) && get_highest(&key) <= to_block {
2323                delete_shard(self, key)?;
2324                deleted = true;
2325            } else {
2326                let removed = block_list.remove_range(0..=to_block);
2327
2328                if block_list.is_empty() {
2329                    delete_shard(self, key)?;
2330                    deleted = true;
2331                } else if removed > 0 {
2332                    put_shard(self, key.clone(), &block_list)?;
2333                    last_remaining = Some((key, block_list));
2334                    updated = true;
2335                } else {
2336                    last_remaining = Some((key, block_list));
2337                }
2338            }
2339        }
2340
2341        if shards_complete &&
2342            let Some((last_key, last_value)) = last_remaining &&
2343            !is_sentinel(&last_key)
2344        {
2345            delete_shard(self, last_key)?;
2346            put_shard(self, create_sentinel(), &last_value)?;
2347            updated = true;
2348        }
2349
2350        if deleted {
2351            Ok(PruneShardOutcome::Deleted)
2352        } else if updated {
2353            Ok(PruneShardOutcome::Updated)
2354        } else {
2355            Ok(PruneShardOutcome::Unchanged)
2356        }
2357    }
2358
2359    /// Prunes account history for the given address, removing blocks <= `to_block`.
2360    ///
2361    /// Mirrors MDBX `prune_shard` semantics. After pruning, the last remaining shard
2362    /// (if any) will have the sentinel key (`u64::MAX`).
2363    pub fn prune_account_history_to(
2364        &mut self,
2365        address: Address,
2366        to_block: BlockNumber,
2367    ) -> ProviderResult<PruneShardOutcome> {
2368        let shards = self.provider.account_history_shards(address)?;
2369        self.prune_history_shards_inner(
2370            shards,
2371            true,
2372            to_block,
2373            |key| key.highest_block_number,
2374            |key| key.highest_block_number == u64::MAX,
2375            |batch, key| batch.delete::<tables::AccountsHistory>(key),
2376            |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2377            || ShardedKey::new(address, u64::MAX),
2378        )
2379    }
2380
2381    /// Prunes account history for multiple addresses in a single iterator pass.
2382    ///
2383    /// This is more efficient than calling [`Self::prune_account_history_to`] repeatedly
2384    /// because it reuses a single raw iterator and skips seeks when the iterator is already
2385    /// positioned correctly (which happens when targets are sorted and adjacent in key order).
2386    ///
2387    /// `targets` MUST be sorted by address and contain each address at most once, for
2388    /// correctness and optimal performance (matches on-disk key order).
2389    pub fn prune_account_history_batch(
2390        &mut self,
2391        targets: &[(Address, BlockNumber)],
2392    ) -> ProviderResult<PrunedIndices> {
2393        if targets.is_empty() {
2394            return Ok(PrunedIndices::default());
2395        }
2396
2397        debug_assert!(
2398            targets.windows(2).all(|w| w[0].0 < w[1].0),
2399            "prune_account_history_batch: targets must be sorted and unique"
2400        );
2401
2402        // ShardedKey<Address> layout: [address: 20][block: 8] = 28 bytes
2403        // The first 20 bytes are the "prefix" that identifies the address
2404        const PREFIX_LEN: usize = 20;
2405
2406        let cf = self.provider.get_cf_handle::<tables::AccountsHistory>()?;
2407        let mut iter = self.provider.0.raw_iterator_cf(cf);
2408        let mut outcomes = PrunedIndices::default();
2409
2410        for (address, to_block) in targets {
2411            // Build the target prefix (first 20 bytes = address)
2412            let start_key = ShardedKey::new(*address, 0u64).encode();
2413            let target_prefix = &start_key[..PREFIX_LEN];
2414
2415            // Check if we need to seek or if the iterator is already positioned correctly.
2416            // After processing the previous target, the iterator is either:
2417            // 1. Positioned at a key with a different prefix (we iterated past our shards)
2418            // 2. Positioned on a later shard of the previous target (we stopped early), whose
2419            //    prefix is below ours because targets are sorted and unique
2420            // 3. Invalid (no more keys)
2421            // If the current key's prefix >= our target prefix, we may be able to skip the seek.
2422            let needs_seek = if iter.valid() {
2423                // If current key's prefix < target prefix, we need to seek forward
2424                // If current key's prefix > target prefix, this target has no shards (skip)
2425                // If current key's prefix == target prefix, we're already positioned
2426                iter.key().is_none_or(|current_key| {
2427                    current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2428                })
2429            } else {
2430                true
2431            };
2432
2433            if needs_seek {
2434                iter.seek(start_key);
2435                iter.status().map_err(|e| {
2436                    ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2437                        message: e.to_string().into(),
2438                        code: -1,
2439                    }))
2440                })?;
2441            }
2442
2443            // Collect the shards for this address that pruning can touch, using raw prefix
2444            // comparison
2445            let mut shards = Vec::new();
2446            let mut shards_complete = true;
2447            while iter.valid() {
2448                let Some(key_bytes) = iter.key() else { break };
2449
2450                // Use raw prefix comparison instead of full decode for the prefix check
2451                let current_prefix = key_bytes.get(..PREFIX_LEN);
2452                if current_prefix != Some(target_prefix) {
2453                    break;
2454                }
2455
2456                // Now decode the full key (we need the block number)
2457                let key = ShardedKey::<Address>::decode(key_bytes)
2458                    .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2459
2460                let Some(value_bytes) = iter.value() else { break };
2461                let value = BlockNumberList::decompress(value_bytes)
2462                    .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2463
2464                let highest = key.highest_block_number;
2465                shards.push((key, value));
2466
2467                iter.next();
2468
2469                // Shards are ordered by their highest block and their contents partition the
2470                // address's history, so this is the last shard holding anything at or below the
2471                // target. Peek past it only to tell whether it was the address's last shard,
2472                // which decides whether a survivor may be re-keyed to the sentinel.
2473                if highest > *to_block {
2474                    shards_complete = iter.key().and_then(|next_key| next_key.get(..PREFIX_LEN)) !=
2475                        Some(target_prefix);
2476                    break;
2477                }
2478            }
2479
2480            // The iterator also goes invalid on a read error, which would otherwise pass a
2481            // truncated shard list off as the key's complete one.
2482            if !iter.valid() {
2483                iter.status().map_err(|e| {
2484                    ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2485                        message: e.to_string().into(),
2486                        code: -1,
2487                    }))
2488                })?;
2489            }
2490
2491            match self.prune_history_shards_inner(
2492                shards,
2493                shards_complete,
2494                *to_block,
2495                |key| key.highest_block_number,
2496                |key| key.highest_block_number == u64::MAX,
2497                |batch, key| batch.delete::<tables::AccountsHistory>(key),
2498                |batch, key, value| batch.put::<tables::AccountsHistory>(key, value),
2499                || ShardedKey::new(*address, u64::MAX),
2500            )? {
2501                PruneShardOutcome::Deleted => outcomes.deleted += 1,
2502                PruneShardOutcome::Updated => outcomes.updated += 1,
2503                PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2504            }
2505        }
2506
2507        Ok(outcomes)
2508    }
2509
2510    /// Prunes storage history for the given address and storage key, removing blocks <=
2511    /// `to_block`.
2512    ///
2513    /// Mirrors MDBX `prune_shard` semantics. After pruning, the last remaining shard
2514    /// (if any) will have the sentinel key (`u64::MAX`).
2515    pub fn prune_storage_history_to(
2516        &mut self,
2517        address: Address,
2518        storage_key: B256,
2519        to_block: BlockNumber,
2520    ) -> ProviderResult<PruneShardOutcome> {
2521        let shards = self.provider.storage_history_shards(address, storage_key)?;
2522        self.prune_history_shards_inner(
2523            shards,
2524            true,
2525            to_block,
2526            |key| key.sharded_key.highest_block_number,
2527            |key| key.sharded_key.highest_block_number == u64::MAX,
2528            |batch, key| batch.delete::<tables::StoragesHistory>(key),
2529            |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2530            || StorageShardedKey::last(address, storage_key),
2531        )
2532    }
2533
2534    /// Prunes storage history for multiple (address, `storage_key`) pairs in a single iterator
2535    /// pass.
2536    ///
2537    /// This is more efficient than calling [`Self::prune_storage_history_to`] repeatedly
2538    /// because it reuses a single raw iterator and skips seeks when the iterator is already
2539    /// positioned correctly (which happens when targets are sorted and adjacent in key order).
2540    ///
2541    /// `targets` MUST be sorted by (address, `storage_key`) and contain each pair at most once,
2542    /// for correctness and optimal performance (matches on-disk key order).
2543    pub fn prune_storage_history_batch(
2544        &mut self,
2545        targets: &[((Address, B256), BlockNumber)],
2546    ) -> ProviderResult<PrunedIndices> {
2547        if targets.is_empty() {
2548            return Ok(PrunedIndices::default());
2549        }
2550
2551        debug_assert!(
2552            targets.windows(2).all(|w| w[0].0 < w[1].0),
2553            "prune_storage_history_batch: targets must be sorted and unique"
2554        );
2555
2556        // StorageShardedKey layout: [address: 20][storage_key: 32][block: 8] = 60 bytes
2557        // The first 52 bytes are the "prefix" that identifies (address, storage_key)
2558        const PREFIX_LEN: usize = 52;
2559
2560        let cf = self.provider.get_cf_handle::<tables::StoragesHistory>()?;
2561        let mut iter = self.provider.0.raw_iterator_cf(cf);
2562        let mut outcomes = PrunedIndices::default();
2563
2564        for ((address, storage_key), to_block) in targets {
2565            // Build the target prefix (first 52 bytes of encoded key)
2566            let start_key = StorageShardedKey::new(*address, *storage_key, 0u64).encode();
2567            let target_prefix = &start_key[..PREFIX_LEN];
2568
2569            // Check if we need to seek or if the iterator is already positioned correctly.
2570            // After processing the previous target, the iterator is either:
2571            // 1. Positioned at a key with a different prefix (we iterated past our shards)
2572            // 2. Positioned on a later shard of the previous target (we stopped early), whose
2573            //    prefix is below ours because targets are sorted and unique
2574            // 3. Invalid (no more keys)
2575            // If the current key's prefix >= our target prefix, we may be able to skip the seek.
2576            let needs_seek = if iter.valid() {
2577                // If current key's prefix < target prefix, we need to seek forward
2578                // If current key's prefix > target prefix, this target has no shards (skip)
2579                // If current key's prefix == target prefix, we're already positioned
2580                iter.key().is_none_or(|current_key| {
2581                    current_key.get(..PREFIX_LEN).is_none_or(|p| p < target_prefix)
2582                })
2583            } else {
2584                true
2585            };
2586
2587            if needs_seek {
2588                iter.seek(start_key);
2589                iter.status().map_err(|e| {
2590                    ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2591                        message: e.to_string().into(),
2592                        code: -1,
2593                    }))
2594                })?;
2595            }
2596
2597            // Collect the shards for this (address, storage_key) pair that pruning can touch,
2598            // using prefix comparison
2599            let mut shards = Vec::new();
2600            let mut shards_complete = true;
2601            while iter.valid() {
2602                let Some(key_bytes) = iter.key() else { break };
2603
2604                // Use raw prefix comparison instead of full decode for the prefix check
2605                let current_prefix = key_bytes.get(..PREFIX_LEN);
2606                if current_prefix != Some(target_prefix) {
2607                    break;
2608                }
2609
2610                // Now decode the full key (we need the block number)
2611                let key = StorageShardedKey::decode(key_bytes)
2612                    .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2613
2614                let Some(value_bytes) = iter.value() else { break };
2615                let value = BlockNumberList::decompress(value_bytes)
2616                    .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
2617
2618                let highest = key.sharded_key.highest_block_number;
2619                shards.push((key, value));
2620
2621                iter.next();
2622
2623                // Shards are ordered by their highest block and their contents partition the
2624                // key's history, so this is the last shard holding anything at or below the
2625                // target. Peek past it only to tell whether it was the key's last shard, which
2626                // decides whether a survivor may be re-keyed to the sentinel.
2627                if highest > *to_block {
2628                    shards_complete = iter.key().and_then(|next_key| next_key.get(..PREFIX_LEN)) !=
2629                        Some(target_prefix);
2630                    break;
2631                }
2632            }
2633
2634            // The iterator also goes invalid on a read error, which would otherwise pass a
2635            // truncated shard list off as the key's complete one.
2636            if !iter.valid() {
2637                iter.status().map_err(|e| {
2638                    ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2639                        message: e.to_string().into(),
2640                        code: -1,
2641                    }))
2642                })?;
2643            }
2644
2645            // Use existing prune_history_shards_inner logic
2646            match self.prune_history_shards_inner(
2647                shards,
2648                shards_complete,
2649                *to_block,
2650                |key| key.sharded_key.highest_block_number,
2651                |key| key.sharded_key.highest_block_number == u64::MAX,
2652                |batch, key| batch.delete::<tables::StoragesHistory>(key),
2653                |batch, key, value| batch.put::<tables::StoragesHistory>(key, value),
2654                || StorageShardedKey::last(*address, *storage_key),
2655            )? {
2656                PruneShardOutcome::Deleted => outcomes.deleted += 1,
2657                PruneShardOutcome::Updated => outcomes.updated += 1,
2658                PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
2659            }
2660        }
2661
2662        Ok(outcomes)
2663    }
2664
2665    /// Unwinds storage history to keep only blocks `<= keep_to`.
2666    ///
2667    /// Handles multi-shard scenarios by:
2668    /// 1. Loading all shards for the `(address, storage_key)` pair
2669    /// 2. Finding the boundary shard containing `keep_to`
2670    /// 3. Deleting all shards after the boundary
2671    /// 4. Truncating the boundary shard to keep only indices `<= keep_to`
2672    /// 5. Ensuring the last shard is keyed with `u64::MAX`
2673    pub fn unwind_storage_history_to(
2674        &mut self,
2675        address: Address,
2676        storage_key: B256,
2677        keep_to: BlockNumber,
2678    ) -> ProviderResult<()> {
2679        let shards = self.provider.storage_history_shards(address, storage_key)?;
2680        if shards.is_empty() {
2681            return Ok(());
2682        }
2683
2684        // Find the first shard that might contain blocks > keep_to.
2685        // A shard is affected if it's the sentinel (u64::MAX) or its highest_block_number > keep_to
2686        let boundary_idx = shards.iter().position(|(key, _)| {
2687            key.sharded_key.highest_block_number == u64::MAX ||
2688                key.sharded_key.highest_block_number > keep_to
2689        });
2690
2691        // Repair path: no shards affected means all blocks <= keep_to, just ensure sentinel exists
2692        let Some(boundary_idx) = boundary_idx else {
2693            let (last_key, last_value) = shards.last().expect("shards is non-empty");
2694            if last_key.sharded_key.highest_block_number != u64::MAX {
2695                self.delete::<tables::StoragesHistory>(last_key.clone())?;
2696                self.put::<tables::StoragesHistory>(
2697                    StorageShardedKey::last(address, storage_key),
2698                    last_value,
2699                )?;
2700            }
2701            return Ok(());
2702        };
2703
2704        // Delete all shards strictly after the boundary (they are entirely > keep_to)
2705        for (key, _) in shards.iter().skip(boundary_idx + 1) {
2706            self.delete::<tables::StoragesHistory>(key.clone())?;
2707        }
2708
2709        // Process the boundary shard: filter out blocks > keep_to
2710        let (boundary_key, boundary_list) = &shards[boundary_idx];
2711
2712        // Delete the boundary shard (we'll either drop it or rewrite at u64::MAX)
2713        self.delete::<tables::StoragesHistory>(boundary_key.clone())?;
2714
2715        // Build truncated list once; check emptiness directly (avoids double iteration)
2716        let new_last =
2717            BlockNumberList::new_pre_sorted(boundary_list.iter().take_while(|&b| b <= keep_to));
2718
2719        if new_last.is_empty() {
2720            // Boundary shard is now empty. Previous shard becomes the last and must be keyed
2721            // u64::MAX.
2722            if boundary_idx == 0 {
2723                // Nothing left for this (address, storage_key) pair
2724                return Ok(());
2725            }
2726
2727            let (prev_key, prev_value) = &shards[boundary_idx - 1];
2728            if prev_key.sharded_key.highest_block_number != u64::MAX {
2729                self.delete::<tables::StoragesHistory>(prev_key.clone())?;
2730                self.put::<tables::StoragesHistory>(
2731                    StorageShardedKey::last(address, storage_key),
2732                    prev_value,
2733                )?;
2734            }
2735            return Ok(());
2736        }
2737
2738        self.put::<tables::StoragesHistory>(
2739            StorageShardedKey::last(address, storage_key),
2740            &new_last,
2741        )?;
2742
2743        Ok(())
2744    }
2745
2746    /// Clears all account history shards for the given address.
2747    ///
2748    /// Used when unwinding from block 0 (i.e., removing all history).
2749    pub fn clear_account_history(&mut self, address: Address) -> ProviderResult<()> {
2750        let shards = self.provider.account_history_shards(address)?;
2751        for (key, _) in shards {
2752            self.delete::<tables::AccountsHistory>(key)?;
2753        }
2754        Ok(())
2755    }
2756
2757    /// Clears all storage history shards for the given `(address, storage_key)` pair.
2758    ///
2759    /// Used when unwinding from block 0 (i.e., removing all history for this storage slot).
2760    pub fn clear_storage_history(
2761        &mut self,
2762        address: Address,
2763        storage_key: B256,
2764    ) -> ProviderResult<()> {
2765        let shards = self.provider.storage_history_shards(address, storage_key)?;
2766        for (key, _) in shards {
2767            self.delete::<tables::StoragesHistory>(key)?;
2768        }
2769        Ok(())
2770    }
2771}
2772
2773/// `RocksDB` transaction wrapper providing MDBX-like semantics.
2774///
2775/// Supports:
2776/// - Read-your-writes: reads see uncommitted writes within the same transaction
2777/// - Atomic commit/rollback
2778/// - Iteration over uncommitted data
2779///
2780/// Note: `Transaction` is `Send` but NOT `Sync`. This wrapper does not implement
2781/// `DbTx`/`DbTxMut` traits directly; use RocksDB-specific methods instead.
2782pub struct RocksTx<'db> {
2783    inner: Transaction<'db, OptimisticTransactionDB>,
2784    provider: &'db RocksDBProvider,
2785}
2786
2787impl fmt::Debug for RocksTx<'_> {
2788    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2789        f.debug_struct("RocksTx").field("provider", &self.provider).finish_non_exhaustive()
2790    }
2791}
2792
2793impl<'db> RocksTx<'db> {
2794    /// Gets a value from the specified table. Sees uncommitted writes in this transaction.
2795    pub fn get<T: Table>(&self, key: T::Key) -> ProviderResult<Option<T::Value>> {
2796        let encoded_key = key.encode();
2797        self.get_encoded::<T>(&encoded_key)
2798    }
2799
2800    /// Gets a value using pre-encoded key. Sees uncommitted writes in this transaction.
2801    pub fn get_encoded<T: Table>(
2802        &self,
2803        key: &<T::Key as Encode>::Encoded,
2804    ) -> ProviderResult<Option<T::Value>> {
2805        let cf = self.provider.get_cf_handle::<T>()?;
2806        let result = self.inner.get_cf(cf, key.as_ref()).map_err(|e| {
2807            ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
2808                message: e.to_string().into(),
2809                code: -1,
2810            }))
2811        })?;
2812
2813        Ok(result.and_then(|value| T::Value::decompress(&value).ok()))
2814    }
2815
2816    /// Puts a value into the specified table.
2817    pub fn put<T: Table>(&self, key: T::Key, value: &T::Value) -> ProviderResult<()> {
2818        let encoded_key = key.encode();
2819        self.put_encoded::<T>(&encoded_key, value)
2820    }
2821
2822    /// Puts a value using pre-encoded key.
2823    pub fn put_encoded<T: Table>(
2824        &self,
2825        key: &<T::Key as Encode>::Encoded,
2826        value: &T::Value,
2827    ) -> ProviderResult<()> {
2828        let cf = self.provider.get_cf_handle::<T>()?;
2829        let mut buf = Vec::new();
2830        let value_bytes = compress_to_buf_or_ref!(buf, value).unwrap_or(&buf);
2831
2832        self.inner.put_cf(cf, key.as_ref(), value_bytes).map_err(|e| {
2833            ProviderError::Database(DatabaseError::Write(Box::new(DatabaseWriteError {
2834                info: DatabaseErrorInfo { message: e.to_string().into(), code: -1 },
2835                operation: DatabaseWriteOperation::PutUpsert,
2836                table_name: T::NAME,
2837                key: key.as_ref().to_vec(),
2838            })))
2839        })
2840    }
2841
2842    /// Deletes a value from the specified table.
2843    pub fn delete<T: Table>(&self, key: T::Key) -> ProviderResult<()> {
2844        let cf = self.provider.get_cf_handle::<T>()?;
2845        self.inner.delete_cf(cf, key.encode().as_ref()).map_err(|e| {
2846            ProviderError::Database(DatabaseError::Delete(DatabaseErrorInfo {
2847                message: e.to_string().into(),
2848                code: -1,
2849            }))
2850        })
2851    }
2852
2853    /// Creates an iterator for the specified table. Sees uncommitted writes in this transaction.
2854    ///
2855    /// Returns an iterator that yields `(encoded_key, compressed_value)` pairs.
2856    pub fn iter<T: Table>(&self) -> ProviderResult<RocksTxIter<'_, T>> {
2857        let cf = self.provider.get_cf_handle::<T>()?;
2858        let iter = self.inner.iterator_cf(cf, IteratorMode::Start);
2859        Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2860    }
2861
2862    /// Creates an iterator starting from the given key (inclusive).
2863    pub fn iter_from<T: Table>(&self, key: T::Key) -> ProviderResult<RocksTxIter<'_, T>> {
2864        let cf = self.provider.get_cf_handle::<T>()?;
2865        let encoded_key = key.encode();
2866        let iter = self
2867            .inner
2868            .iterator_cf(cf, IteratorMode::From(encoded_key.as_ref(), rocksdb::Direction::Forward));
2869        Ok(RocksTxIter { inner: iter, _marker: std::marker::PhantomData })
2870    }
2871
2872    /// Commits the transaction, persisting all changes.
2873    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2874    pub fn commit(self) -> ProviderResult<()> {
2875        self.inner.commit().map_err(|e| {
2876            ProviderError::Database(DatabaseError::Commit(DatabaseErrorInfo {
2877                message: e.to_string().into(),
2878                code: -1,
2879            }))
2880        })
2881    }
2882
2883    /// Rolls back the transaction, discarding all changes.
2884    #[instrument(level = "debug", target = "providers::rocksdb", skip_all)]
2885    pub fn rollback(self) -> ProviderResult<()> {
2886        self.inner.rollback().map_err(|e| {
2887            ProviderError::Database(DatabaseError::Other(format!("rollback failed: {e}")))
2888        })
2889    }
2890}
2891
2892/// Wrapper enum for `RocksDB` iterators that works in both read-write and read-only modes.
2893enum RocksDBIterEnum<'db> {
2894    /// Iterator from read-write `OptimisticTransactionDB`.
2895    ReadWrite(rocksdb::DBIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2896    /// Iterator from read-only `DB`.
2897    ReadOnly(rocksdb::DBIteratorWithThreadMode<'db, DB>),
2898}
2899
2900impl Iterator for RocksDBIterEnum<'_> {
2901    type Item = Result<(Box<[u8]>, Box<[u8]>), rocksdb::Error>;
2902
2903    fn next(&mut self) -> Option<Self::Item> {
2904        match self {
2905            Self::ReadWrite(iter) => iter.next(),
2906            Self::ReadOnly(iter) => iter.next(),
2907        }
2908    }
2909}
2910
2911/// Wrapper enum for raw `RocksDB` iterators that works in both read-write and read-only modes.
2912///
2913/// Unlike [`RocksDBIterEnum`], raw iterators expose `seek()` for efficient repositioning
2914/// without reinitializing the iterator.
2915enum RocksDBRawIterEnum<'db> {
2916    /// Raw iterator from read-write `OptimisticTransactionDB`.
2917    ReadWrite(DBRawIteratorWithThreadMode<'db, OptimisticTransactionDB>),
2918    /// Raw iterator from read-only `DB`.
2919    ReadOnly(DBRawIteratorWithThreadMode<'db, DB>),
2920}
2921
2922impl RocksDBRawIterEnum<'_> {
2923    /// Positions the iterator at the first key >= `key`.
2924    fn seek(&mut self, key: impl AsRef<[u8]>) {
2925        match self {
2926            Self::ReadWrite(iter) => iter.seek(key),
2927            Self::ReadOnly(iter) => iter.seek(key),
2928        }
2929    }
2930
2931    /// Returns true if the iterator is positioned at a valid key-value pair.
2932    fn valid(&self) -> bool {
2933        match self {
2934            Self::ReadWrite(iter) => iter.valid(),
2935            Self::ReadOnly(iter) => iter.valid(),
2936        }
2937    }
2938
2939    /// Returns the current key, if valid.
2940    fn key(&self) -> Option<&[u8]> {
2941        match self {
2942            Self::ReadWrite(iter) => iter.key(),
2943            Self::ReadOnly(iter) => iter.key(),
2944        }
2945    }
2946
2947    /// Returns the current value, if valid.
2948    fn value(&self) -> Option<&[u8]> {
2949        match self {
2950            Self::ReadWrite(iter) => iter.value(),
2951            Self::ReadOnly(iter) => iter.value(),
2952        }
2953    }
2954
2955    /// Advances the iterator to the next key.
2956    fn next(&mut self) {
2957        match self {
2958            Self::ReadWrite(iter) => iter.next(),
2959            Self::ReadOnly(iter) => iter.next(),
2960        }
2961    }
2962
2963    /// Moves the iterator to the previous key.
2964    fn prev(&mut self) {
2965        match self {
2966            Self::ReadWrite(iter) => iter.prev(),
2967            Self::ReadOnly(iter) => iter.prev(),
2968        }
2969    }
2970
2971    /// Returns the status of the iterator.
2972    fn status(&self) -> Result<(), rocksdb::Error> {
2973        match self {
2974            Self::ReadWrite(iter) => iter.status(),
2975            Self::ReadOnly(iter) => iter.status(),
2976        }
2977    }
2978}
2979
2980/// Iterator over a `RocksDB` table (non-transactional).
2981///
2982/// Yields decoded `(Key, Value)` pairs in key order.
2983pub struct RocksDBIter<'db, T: Table> {
2984    inner: RocksDBIterEnum<'db>,
2985    _marker: std::marker::PhantomData<T>,
2986}
2987
2988impl<T: Table> fmt::Debug for RocksDBIter<'_, T> {
2989    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2990        f.debug_struct("RocksDBIter").field("table", &T::NAME).finish_non_exhaustive()
2991    }
2992}
2993
2994impl<T: Table> Iterator for RocksDBIter<'_, T> {
2995    type Item = ProviderResult<(T::Key, T::Value)>;
2996
2997    fn next(&mut self) -> Option<Self::Item> {
2998        Some(decode_iter_item::<T>(self.inner.next()?))
2999    }
3000}
3001
3002/// Raw iterator over a `RocksDB` table (non-transactional).
3003///
3004/// Yields raw `(key_bytes, value_bytes)` pairs without decoding.
3005pub struct RocksDBRawIter<'db> {
3006    inner: RocksDBIterEnum<'db>,
3007}
3008
3009impl fmt::Debug for RocksDBRawIter<'_> {
3010    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3011        f.debug_struct("RocksDBRawIter").finish_non_exhaustive()
3012    }
3013}
3014
3015impl Iterator for RocksDBRawIter<'_> {
3016    type Item = ProviderResult<(Box<[u8]>, Box<[u8]>)>;
3017
3018    fn next(&mut self) -> Option<Self::Item> {
3019        match self.inner.next()? {
3020            Ok(kv) => Some(Ok(kv)),
3021            Err(e) => Some(Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
3022                message: e.to_string().into(),
3023                code: -1,
3024            })))),
3025        }
3026    }
3027}
3028
3029/// Raw key iterator over a `RocksDB` table (non-transactional).
3030pub(crate) struct RocksDBRawKeyIter<'db> {
3031    inner: RocksDBRawIterEnum<'db>,
3032}
3033
3034impl fmt::Debug for RocksDBRawKeyIter<'_> {
3035    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3036        f.debug_struct("RocksDBRawKeyIter").finish_non_exhaustive()
3037    }
3038}
3039
3040impl Iterator for RocksDBRawKeyIter<'_> {
3041    type Item = ProviderResult<Box<[u8]>>;
3042
3043    fn next(&mut self) -> Option<Self::Item> {
3044        if !self.inner.valid() {
3045            return self.inner.status().err().map(|e| {
3046                Err(ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
3047                    message: e.to_string().into(),
3048                    code: -1,
3049                })))
3050            })
3051        }
3052
3053        let Some(key) = self.inner.key() else {
3054            return Some(Err(ProviderError::Database(DatabaseError::Decode)))
3055        };
3056        let key = Box::from(key);
3057        self.inner.next();
3058        Some(Ok(key))
3059    }
3060}
3061
3062/// Iterator over a `RocksDB` table within a transaction.
3063///
3064/// Yields decoded `(Key, Value)` pairs. Sees uncommitted writes.
3065pub struct RocksTxIter<'tx, T: Table> {
3066    inner: rocksdb::DBIteratorWithThreadMode<'tx, Transaction<'tx, OptimisticTransactionDB>>,
3067    _marker: std::marker::PhantomData<T>,
3068}
3069
3070impl<T: Table> fmt::Debug for RocksTxIter<'_, T> {
3071    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3072        f.debug_struct("RocksTxIter").field("table", &T::NAME).finish_non_exhaustive()
3073    }
3074}
3075
3076impl<T: Table> Iterator for RocksTxIter<'_, T> {
3077    type Item = ProviderResult<(T::Key, T::Value)>;
3078
3079    fn next(&mut self) -> Option<Self::Item> {
3080        Some(decode_iter_item::<T>(self.inner.next()?))
3081    }
3082}
3083
3084/// Decodes a raw key-value pair from a `RocksDB` iterator into typed table entries.
3085///
3086/// Handles both error propagation from the underlying iterator and
3087/// decoding/decompression of the key and value bytes.
3088fn decode_iter_item<T: Table>(result: RawKVResult) -> ProviderResult<(T::Key, T::Value)> {
3089    let (key_bytes, value_bytes) = result.map_err(|e| {
3090        ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo {
3091            message: e.to_string().into(),
3092            code: -1,
3093        }))
3094    })?;
3095
3096    let key = <T::Key as reth_db_api::table::Decode>::decode(&key_bytes)
3097        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
3098
3099    let value = T::Value::decompress(&value_bytes)
3100        .map_err(|_| ProviderError::Database(DatabaseError::Decode))?;
3101
3102    Ok((key, value))
3103}
3104
3105/// Converts Reth's [`LogLevel`] to `RocksDB`'s [`rocksdb::LogLevel`].
3106const fn convert_log_level(level: LogLevel) -> rocksdb::LogLevel {
3107    match level {
3108        LogLevel::Fatal => rocksdb::LogLevel::Fatal,
3109        LogLevel::Error => rocksdb::LogLevel::Error,
3110        LogLevel::Warn => rocksdb::LogLevel::Warn,
3111        LogLevel::Notice | LogLevel::Verbose => rocksdb::LogLevel::Info,
3112        LogLevel::Debug | LogLevel::Trace | LogLevel::Extra => rocksdb::LogLevel::Debug,
3113    }
3114}
3115
3116/// Selects the `RocksDB` `max_open_files` setting from the current file descriptor limit
3117/// balancing performance vs. compatibility.
3118///
3119/// A mature database can use tens of thousands of file descriptors. With a
3120/// finite `max_open_files` value, `RocksDB` performs additional table-cache checks
3121/// even when enough capacity is available. The value `-1` keeps all table files
3122/// open and avoids this overhead.
3123///
3124/// During normal startup, Reth raises the soft file descriptor limit to the
3125/// system hard limit before it opens `RocksDB`. Therefore, we expect the current
3126/// limit to be the hard limit. A survey of modern Linux distributions shows
3127/// common default limits of `1024:524288`.
3128///
3129/// We use `-1` only when the current limit is above a conservative threshold.
3130/// This keeps enough file descriptors for other parts of the process. When the
3131/// limit is lower or cannot be read, we use a stricter fixed limit.
3132fn select_max_open_files() -> i32 {
3133    let file_descriptor_limit = current_file_descriptor_limit();
3134    let max_open_files = max_open_files_for_limit(file_descriptor_limit);
3135
3136    if max_open_files == LIMITED_MAX_OPEN_FILES {
3137        tracing::warn!(
3138            target: "providers::rocksdb",
3139            ?file_descriptor_limit,
3140            threshold = HIGH_FILE_DESCRIPTOR_LIMIT,
3141            max_open_files,
3142            "RocksDB will not keep all files open; performance may be reduced"
3143        );
3144    }
3145
3146    max_open_files
3147}
3148
3149const fn max_open_files_for_limit(file_descriptor_limit: Option<u64>) -> i32 {
3150    match file_descriptor_limit {
3151        Some(limit) if limit >= HIGH_FILE_DESCRIPTOR_LIMIT => KEEP_ALL_FILES_OPEN,
3152        _ => LIMITED_MAX_OPEN_FILES,
3153    }
3154}
3155
3156#[cfg(unix)]
3157#[allow(clippy::useless_conversion)]
3158fn current_file_descriptor_limit() -> Option<u64> {
3159    let mut limit = libc::rlimit { rlim_cur: 0, rlim_max: 0 };
3160    // SAFETY: `limit` points to initialized writable memory for the kernel to populate.
3161    let result = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) };
3162    (result == 0).then(|| limit.rlim_cur.into())
3163}
3164
3165#[cfg(not(unix))]
3166const fn current_file_descriptor_limit() -> Option<u64> {
3167    None
3168}
3169
3170#[cfg(test)]
3171mod tests {
3172    use super::*;
3173    use crate::providers::HistoryInfo;
3174    use alloy_primitives::{Address, Bytes, TxHash, B256};
3175    use reth_db_api::{
3176        models::{
3177            sharded_key::{ShardedKey, NUM_OF_INDICES_IN_SHARD},
3178            storage_sharded_key::StorageShardedKey,
3179            IntegerList,
3180        },
3181        table::Table,
3182        tables,
3183    };
3184    use rocksdb::{
3185        perf::{set_perf_stats, PerfContext, PerfMetric, PerfStatsLevel},
3186        statistics::{Histogram, Ticker},
3187    };
3188    use tempfile::TempDir;
3189
3190    /// Number of deleted keys `RocksDB` skipped on this thread while running `f`.
3191    fn deleted_keys_skipped<R>(f: impl FnOnce() -> R) -> (R, u64) {
3192        set_perf_stats(PerfStatsLevel::EnableCount);
3193        let mut ctx = PerfContext::default();
3194        ctx.reset();
3195        let result = f();
3196        let skipped = ctx.metric(PerfMetric::InternalDeleteSkippedCount);
3197        set_perf_stats(PerfStatsLevel::Disable);
3198        (result, skipped)
3199    }
3200
3201    fn address_at(i: u64) -> Address {
3202        let mut bytes = [0u8; 20];
3203        bytes[12..].copy_from_slice(&i.to_be_bytes());
3204        Address::from(bytes)
3205    }
3206
3207    /// Looking up the shards of an address whose history was already removed must not scan the
3208    /// deleted history of every address after it. This happens when an unwind (or the startup
3209    /// heal) removes the history of a large state import: each address's shards are looked up
3210    /// in turn, so an unbounded scan makes the whole unwind quadratic in the number of accounts.
3211    #[test]
3212    fn test_history_shards_lookup_skips_only_own_deleted_keys() {
3213        const ADDRESSES: u64 = 5_000;
3214
3215        let temp_dir = TempDir::new().unwrap();
3216        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3217        let storage_key = B256::with_last_byte(1);
3218        let list = BlockNumberList::new_pre_sorted([1]);
3219
3220        // History for many addresses, then a live address after all of them.
3221        let mut batch = provider.batch();
3222        for i in 0..ADDRESSES {
3223            let address = address_at(i);
3224            batch
3225                .put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &list)
3226                .unwrap();
3227            batch
3228                .put::<tables::StoragesHistory>(
3229                    StorageShardedKey::new(address, storage_key, u64::MAX),
3230                    &list,
3231                )
3232                .unwrap();
3233        }
3234        let live = Address::repeat_byte(0xff);
3235        batch.put::<tables::AccountsHistory>(ShardedKey::new(live, u64::MAX), &list).unwrap();
3236        batch
3237            .put::<tables::StoragesHistory>(
3238                StorageShardedKey::new(live, storage_key, u64::MAX),
3239                &list,
3240            )
3241            .unwrap();
3242        batch.commit().unwrap();
3243
3244        // Remove the history of all but the live address, as an unwind would.
3245        let mut batch = provider.batch();
3246        for i in 0..ADDRESSES {
3247            batch.clear_account_history(address_at(i)).unwrap();
3248            batch.clear_storage_history(address_at(i), storage_key).unwrap();
3249        }
3250        batch.commit().unwrap();
3251
3252        let (shards, skipped) =
3253            deleted_keys_skipped(|| provider.account_history_shards(address_at(0)));
3254        assert!(shards.unwrap().is_empty());
3255        assert!(skipped <= 1, "account lookup skipped {skipped} deleted keys");
3256
3257        let (shards, skipped) =
3258            deleted_keys_skipped(|| provider.storage_history_shards(address_at(0), storage_key));
3259        assert!(shards.unwrap().is_empty());
3260        assert!(skipped <= 1, "storage lookup skipped {skipped} deleted keys");
3261
3262        // The live address after the deleted range is still found.
3263        assert_eq!(provider.account_history_shards(live).unwrap().len(), 1);
3264        assert_eq!(provider.storage_history_shards(live, storage_key).unwrap().len(), 1);
3265    }
3266
3267    #[test]
3268    fn max_open_files_adapts_to_file_descriptor_limit() {
3269        assert_eq!(max_open_files_for_limit(Some(HIGH_FILE_DESCRIPTOR_LIMIT)), KEEP_ALL_FILES_OPEN);
3270        assert_eq!(
3271            max_open_files_for_limit(Some(HIGH_FILE_DESCRIPTOR_LIMIT - 1)),
3272            LIMITED_MAX_OPEN_FILES
3273        );
3274        assert_eq!(max_open_files_for_limit(None), LIMITED_MAX_OPEN_FILES);
3275    }
3276
3277    #[test]
3278    fn test_with_default_tables_registers_required_column_families() {
3279        let temp_dir = TempDir::new().unwrap();
3280
3281        // Build with default tables
3282        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3283
3284        // Should be able to write/read TransactionHashNumbers
3285        let tx_hash = TxHash::from(B256::repeat_byte(1u8));
3286        provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3287        assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3288
3289        // Should be able to write/read AccountsHistory
3290        let key = ShardedKey::new(Address::ZERO, 100);
3291        let value = IntegerList::default();
3292        provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
3293        assert!(provider.get::<tables::AccountsHistory>(key).unwrap().is_some());
3294
3295        // Should be able to write/read StoragesHistory
3296        let key = StorageShardedKey::new(Address::ZERO, B256::ZERO, 100);
3297        provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
3298        assert!(provider.get::<tables::StoragesHistory>(key).unwrap().is_some());
3299
3300        let bal_key =
3301            reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3302        let bal_value =
3303            reth_db_api::models::StoredBlockAccessList::new(Bytes::from_static(&[0xc0]));
3304        provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3305        assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
3306        provider
3307            .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
3308            .unwrap();
3309        assert_eq!(
3310            provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
3311            Some(bal_key.number())
3312        );
3313    }
3314
3315    #[test]
3316    fn block_access_lists_store_large_payloads_in_blob_files() {
3317        let temp_dir = TempDir::new().unwrap();
3318        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3319        let bal_key =
3320            reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3321        let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
3322            0;
3323            DEFAULT_BAL_MIN_BLOB_SIZE as usize +
3324                1
3325        ]));
3326
3327        provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3328        provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3329
3330        let has_blob_file = std::fs::read_dir(temp_dir.path()).unwrap().any(|entry| {
3331            entry.unwrap().path().extension().is_some_and(|extension| extension == "blob")
3332        });
3333        assert!(has_blob_file);
3334    }
3335
3336    #[derive(Debug)]
3337    struct TestTable;
3338
3339    impl Table for TestTable {
3340        const NAME: &'static str = "TestTable";
3341        const DUPSORT: bool = false;
3342        type Key = u64;
3343        type Value = Vec<u8>;
3344    }
3345
3346    #[test]
3347    fn test_reopens_with_unknown_column_family() {
3348        let temp_dir = TempDir::new().unwrap();
3349        let value = b"test_value".to_vec();
3350
3351        let provider = RocksDBBuilder::new(temp_dir.path())
3352            .with_default_tables()
3353            .with_table::<TestTable>()
3354            .build()
3355            .unwrap();
3356        provider.put::<TestTable>(42, &value).unwrap();
3357        drop(provider);
3358
3359        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3360        assert_eq!(provider.get::<TestTable>(42).unwrap(), Some(value));
3361    }
3362
3363    #[test]
3364    fn test_reopens_blob_column_family_with_legacy_table_set() {
3365        let temp_dir = TempDir::new().unwrap();
3366        let bal_key =
3367            reth_db_api::models::StoredBlockAccessListKey::new(1, B256::with_last_byte(1));
3368        let bal_value = reth_db_api::models::StoredBlockAccessList::new(Bytes::from(vec![
3369            0;
3370            DEFAULT_BAL_MIN_BLOB_SIZE as usize +
3371                1
3372        ]));
3373
3374        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3375        provider.put::<tables::BlockAccessLists>(bal_key, &bal_value).unwrap();
3376        provider
3377            .put::<tables::BlockAccessListBlockNumbers>(bal_key.hash(), &bal_key.number())
3378            .unwrap();
3379        provider.flush(&[tables::BlockAccessLists::NAME]).unwrap();
3380        drop(provider);
3381
3382        let provider = RocksDBBuilder::new(temp_dir.path())
3383            .with_table::<tables::TransactionHashNumbers>()
3384            .with_table::<tables::AccountsHistory>()
3385            .with_table::<tables::StoragesHistory>()
3386            .build()
3387            .unwrap();
3388        assert_eq!(provider.get::<tables::BlockAccessLists>(bal_key).unwrap(), Some(bal_value));
3389        assert_eq!(
3390            provider.get::<tables::BlockAccessListBlockNumbers>(bal_key.hash()).unwrap(),
3391            Some(bal_key.number())
3392        );
3393    }
3394
3395    #[test]
3396    fn test_basic_operations() {
3397        let temp_dir = TempDir::new().unwrap();
3398
3399        let provider = RocksDBBuilder::new(temp_dir.path())
3400            .with_table::<TestTable>() // Type-safe!
3401            .build()
3402            .unwrap();
3403
3404        let key = 42u64;
3405        let value = b"test_value".to_vec();
3406
3407        // Test write
3408        provider.put::<TestTable>(key, &value).unwrap();
3409
3410        // Test read
3411        let result = provider.get::<TestTable>(key).unwrap();
3412        assert_eq!(result, Some(value));
3413
3414        // Test delete
3415        provider.delete::<TestTable>(key).unwrap();
3416
3417        // Verify deletion
3418        assert_eq!(provider.get::<TestTable>(key).unwrap(), None);
3419    }
3420
3421    #[test]
3422    fn test_batch_operations() {
3423        let temp_dir = TempDir::new().unwrap();
3424        let provider =
3425            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3426
3427        // Write multiple entries in a batch
3428        provider
3429            .write_batch(|batch| {
3430                for i in 0..10u64 {
3431                    let value = format!("value_{i}").into_bytes();
3432                    batch.put::<TestTable>(i, &value)?;
3433                }
3434                Ok(())
3435            })
3436            .unwrap();
3437
3438        // Read all entries
3439        for i in 0..10u64 {
3440            let value = format!("value_{i}").into_bytes();
3441            assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3442        }
3443
3444        // Delete all entries in a batch
3445        provider
3446            .write_batch(|batch| {
3447                for i in 0..10u64 {
3448                    batch.delete::<TestTable>(i)?;
3449                }
3450                Ok(())
3451            })
3452            .unwrap();
3453
3454        // Verify all deleted
3455        for i in 0..10u64 {
3456            assert_eq!(provider.get::<TestTable>(i).unwrap(), None);
3457        }
3458    }
3459
3460    #[test]
3461    fn test_with_real_table() {
3462        let temp_dir = TempDir::new().unwrap();
3463        let provider = RocksDBBuilder::new(temp_dir.path())
3464            .with_table::<tables::TransactionHashNumbers>()
3465            .with_metrics()
3466            .build()
3467            .unwrap();
3468
3469        let tx_hash = TxHash::from(B256::repeat_byte(1u8));
3470
3471        // Insert and retrieve
3472        provider.put::<tables::TransactionHashNumbers>(tx_hash, &100).unwrap();
3473        assert_eq!(provider.get::<tables::TransactionHashNumbers>(tx_hash).unwrap(), Some(100));
3474
3475        // Batch insert multiple transactions
3476        provider
3477            .write_batch(|batch| {
3478                for i in 0..10u64 {
3479                    let hash = TxHash::from(B256::repeat_byte(i as u8));
3480                    let value = i * 100;
3481                    batch.put::<tables::TransactionHashNumbers>(hash, &value)?;
3482                }
3483                Ok(())
3484            })
3485            .unwrap();
3486
3487        // Verify batch insertions
3488        for i in 0..10u64 {
3489            let hash = TxHash::from(B256::repeat_byte(i as u8));
3490            assert_eq!(
3491                provider.get::<tables::TransactionHashNumbers>(hash).unwrap(),
3492                Some(i * 100)
3493            );
3494        }
3495    }
3496    #[test]
3497    fn test_statistics_enabled() {
3498        let temp_dir = TempDir::new().unwrap();
3499        // Just verify that building with statistics doesn't panic
3500        let provider = RocksDBBuilder::new(temp_dir.path())
3501            .with_table::<TestTable>()
3502            .with_statistics()
3503            .build()
3504            .unwrap();
3505
3506        // Do operations - data should be immediately readable with OptimisticTransactionDB
3507        for i in 0..10 {
3508            let value = vec![i as u8];
3509            provider.put::<TestTable>(i, &value).unwrap();
3510            // Verify write is visible
3511            assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3512        }
3513    }
3514
3515    /// Guards the statistics level: the counters must keep working, the per-operation timer
3516    /// histograms must stay empty. `rocksdb`'s `StatsLevel` discriminants do not line up with the
3517    /// levels the C API defines, so the variant name alone does not tell us what was selected.
3518    #[test]
3519    fn test_statistics_level_skips_timers() {
3520        let temp_dir = TempDir::new().unwrap();
3521        let cache = Cache::new_lru_cache(1 << 20);
3522        let options = RocksDBBuilder::default_options(rocksdb::LogLevel::Info, &cache, true);
3523
3524        let db = DB::open(&options, temp_dir.path()).unwrap();
3525        for i in 0..10u8 {
3526            db.put([i], [i]).unwrap();
3527            assert_eq!(db.get([i]).unwrap(), Some(vec![i]));
3528        }
3529
3530        assert!(options.get_ticker_count(Ticker::NumberKeysRead) > 0);
3531        assert_eq!(options.get_histogram_data(Histogram::DbGet).count(), 0);
3532    }
3533
3534    #[test]
3535    fn test_data_persistence() {
3536        let temp_dir = TempDir::new().unwrap();
3537        let provider =
3538            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3539
3540        // Insert data - OptimisticTransactionDB writes are immediately visible
3541        let value = vec![42u8; 1000];
3542        for i in 0..100 {
3543            provider.put::<TestTable>(i, &value).unwrap();
3544        }
3545
3546        // Verify data is readable
3547        for i in 0..100 {
3548            assert!(provider.get::<TestTable>(i).unwrap().is_some(), "Data should be readable");
3549        }
3550    }
3551
3552    #[test]
3553    fn test_transaction_read_your_writes() {
3554        let temp_dir = TempDir::new().unwrap();
3555        let provider =
3556            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3557
3558        // Create a transaction
3559        let tx = provider.tx();
3560
3561        // Write data within the transaction
3562        let key = 42u64;
3563        let value = b"test_value".to_vec();
3564        tx.put::<TestTable>(key, &value).unwrap();
3565
3566        // Read-your-writes: should see uncommitted data in same transaction
3567        let result = tx.get::<TestTable>(key).unwrap();
3568        assert_eq!(
3569            result,
3570            Some(value.clone()),
3571            "Transaction should see its own uncommitted writes"
3572        );
3573
3574        // Data should NOT be visible via provider (outside transaction)
3575        let provider_result = provider.get::<TestTable>(key).unwrap();
3576        assert_eq!(provider_result, None, "Uncommitted data should not be visible outside tx");
3577
3578        // Commit the transaction
3579        tx.commit().unwrap();
3580
3581        // Now data should be visible via provider
3582        let committed_result = provider.get::<TestTable>(key).unwrap();
3583        assert_eq!(committed_result, Some(value), "Committed data should be visible");
3584    }
3585
3586    #[test]
3587    fn test_transaction_rollback() {
3588        let temp_dir = TempDir::new().unwrap();
3589        let provider =
3590            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3591
3592        // First, put some initial data
3593        let key = 100u64;
3594        let initial_value = b"initial".to_vec();
3595        provider.put::<TestTable>(key, &initial_value).unwrap();
3596
3597        // Create a transaction and modify data
3598        let tx = provider.tx();
3599        let new_value = b"modified".to_vec();
3600        tx.put::<TestTable>(key, &new_value).unwrap();
3601
3602        // Verify modification is visible within transaction
3603        assert_eq!(tx.get::<TestTable>(key).unwrap(), Some(new_value));
3604
3605        // Rollback instead of commit
3606        tx.rollback().unwrap();
3607
3608        // Data should be unchanged (initial value)
3609        let result = provider.get::<TestTable>(key).unwrap();
3610        assert_eq!(result, Some(initial_value), "Rollback should preserve original data");
3611    }
3612
3613    #[test]
3614    fn test_transaction_iterator() {
3615        let temp_dir = TempDir::new().unwrap();
3616        let provider =
3617            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3618
3619        // Create a transaction
3620        let tx = provider.tx();
3621
3622        // Write multiple entries
3623        for i in 0..5u64 {
3624            let value = format!("value_{i}").into_bytes();
3625            tx.put::<TestTable>(i, &value).unwrap();
3626        }
3627
3628        // Iterate - should see uncommitted writes
3629        let mut count = 0;
3630        for result in tx.iter::<TestTable>().unwrap() {
3631            let (key, value) = result.unwrap();
3632            assert_eq!(value, format!("value_{key}").into_bytes());
3633            count += 1;
3634        }
3635        assert_eq!(count, 5, "Iterator should see all uncommitted writes");
3636
3637        // Commit
3638        tx.commit().unwrap();
3639    }
3640
3641    #[test]
3642    fn test_batch_manual_commit() {
3643        let temp_dir = TempDir::new().unwrap();
3644        let provider =
3645            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3646
3647        // Create a batch via provider.batch()
3648        let mut batch = provider.batch();
3649
3650        // Add entries
3651        for i in 0..10u64 {
3652            let value = format!("batch_value_{i}").into_bytes();
3653            batch.put::<TestTable>(i, &value).unwrap();
3654        }
3655
3656        // Verify len/is_empty
3657        assert_eq!(batch.len(), 10);
3658        assert!(!batch.is_empty());
3659
3660        // Data should NOT be visible before commit
3661        assert_eq!(provider.get::<TestTable>(0).unwrap(), None);
3662
3663        // Commit the batch
3664        batch.commit().unwrap();
3665
3666        // Now data should be visible
3667        for i in 0..10u64 {
3668            let value = format!("batch_value_{i}").into_bytes();
3669            assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
3670        }
3671    }
3672
3673    #[test]
3674    fn test_first_and_last_entry() {
3675        let temp_dir = TempDir::new().unwrap();
3676        let provider =
3677            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
3678
3679        // Empty table should return None for both
3680        assert_eq!(provider.first::<TestTable>().unwrap(), None);
3681        assert_eq!(provider.last::<TestTable>().unwrap(), None);
3682
3683        // Insert some entries
3684        provider.put::<TestTable>(10, &b"value_10".to_vec()).unwrap();
3685        provider.put::<TestTable>(20, &b"value_20".to_vec()).unwrap();
3686        provider.put::<TestTable>(5, &b"value_5".to_vec()).unwrap();
3687
3688        // First should return the smallest key
3689        let first = provider.first::<TestTable>().unwrap();
3690        assert_eq!(first, Some((5, b"value_5".to_vec())));
3691
3692        // Last should return the largest key
3693        let last = provider.last::<TestTable>().unwrap();
3694        assert_eq!(last, Some((20, b"value_20".to_vec())));
3695    }
3696
3697    #[test]
3698    fn test_owned_history_snapshot_outlives_provider() {
3699        let temp_dir = TempDir::new().unwrap();
3700        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3701        let address = Address::repeat_byte(0x42);
3702        let slot = B256::repeat_byte(0x43);
3703        let chunk = IntegerList::new([100, 200, 300]).unwrap();
3704        provider
3705            .put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &chunk)
3706            .unwrap();
3707        provider
3708            .put::<tables::StoragesHistory>(StorageShardedKey::new(address, slot, u64::MAX), &chunk)
3709            .unwrap();
3710
3711        let owned = provider.owned_snapshot();
3712        let snapshot = owned.as_snapshot();
3713        assert_eq!(
3714            snapshot.account_history_info(address, 200, None, u64::MAX).unwrap(),
3715            HistoryInfo::InChangeset(200)
3716        );
3717        assert_eq!(
3718            snapshot.storage_history_info(address, slot, 200, None, u64::MAX).unwrap(),
3719            HistoryInfo::InChangeset(200)
3720        );
3721        drop(provider);
3722
3723        // Move the owner with populated iterator caches, then seek both forwards and backwards.
3724        std::thread::spawn(move || {
3725            let snapshot = owned.as_snapshot();
3726            for (block, expected) in [
3727                (400, HistoryInfo::InPlainState),
3728                (50, HistoryInfo::NotYetWritten),
3729                (200, HistoryInfo::InChangeset(200)),
3730            ] {
3731                assert_eq!(
3732                    snapshot.account_history_info(address, block, None, u64::MAX).unwrap(),
3733                    expected
3734                );
3735                assert_eq!(
3736                    snapshot.storage_history_info(address, slot, block, None, u64::MAX).unwrap(),
3737                    expected
3738                );
3739            }
3740        })
3741        .join()
3742        .unwrap();
3743    }
3744
3745    #[test]
3746    fn test_history_snapshot_cached_and_private_iterators_keep_same_view() {
3747        let temp_dir = TempDir::new().unwrap();
3748        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3749        let address = Address::repeat_byte(0x42);
3750        let slot = B256::repeat_byte(0x43);
3751        let account_key = ShardedKey::new(address, u64::MAX);
3752        let storage_key = StorageShardedKey::new(address, slot, u64::MAX);
3753        let chunk = IntegerList::new([100, 200]).unwrap();
3754        provider.put::<tables::AccountsHistory>(account_key.clone(), &chunk).unwrap();
3755        provider.put::<tables::StoragesHistory>(storage_key.clone(), &chunk).unwrap();
3756
3757        let owned = provider.owned_snapshot();
3758        let snapshot = owned.as_snapshot();
3759        let check = |snapshot: &RocksReadSnapshot<'_>, expected| {
3760            assert_eq!(
3761                snapshot.account_history_info(address, 125, None, u64::MAX).unwrap(),
3762                expected
3763            );
3764            assert_eq!(
3765                snapshot.storage_history_info(address, slot, 125, None, u64::MAX).unwrap(),
3766                expected
3767            );
3768        };
3769        check(snapshot, HistoryInfo::InChangeset(200));
3770
3771        let updated = IntegerList::new([100, 150, 200]).unwrap();
3772        provider.put::<tables::AccountsHistory>(account_key, &updated).unwrap();
3773        provider.put::<tables::StoragesHistory>(storage_key, &updated).unwrap();
3774        check(&provider.snapshot(), HistoryInfo::InChangeset(150));
3775        check(snapshot, HistoryInfo::InChangeset(200));
3776
3777        // Force another thread to use private iterators after the database has changed.
3778        let _accounts = snapshot.accounts_history_iter.lock();
3779        let _storages = snapshot.storages_history_iter.lock();
3780        std::thread::scope(|scope| {
3781            scope.spawn(|| check(snapshot, HistoryInfo::InChangeset(200))).join().unwrap();
3782        });
3783    }
3784
3785    /// Tests the edge case where block < `lowest_available_block_number`.
3786    ///
3787    /// State queries reject this before the `RocksDB` lookup, so this verifies the low-level
3788    /// behavior directly.
3789    #[test]
3790    fn test_account_history_info_pruned_before_first_entry() {
3791        let temp_dir = TempDir::new().unwrap();
3792        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3793
3794        let address = Address::repeat_byte(0x42);
3795
3796        // Create a single shard starting at block 100
3797        let chunk = IntegerList::new([100, 200, 300]).unwrap();
3798        let shard_key = ShardedKey::new(address, u64::MAX);
3799        provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3800
3801        // Query for block 50 with lowest_available_block_number = 100
3802        // This simulates a pruned state where data before block 100 is not available.
3803        // Since we're before the first write AND pruning boundary is set, we need to
3804        // check the changeset at the first write block.
3805        let result =
3806            provider.snapshot().account_history_info(address, 50, Some(100), u64::MAX).unwrap();
3807        assert_eq!(result, HistoryInfo::InChangeset(100));
3808    }
3809
3810    /// Verifies that a read-only (secondary) provider can catch up with primary writes.
3811    #[test]
3812    fn test_account_history_info_read_only_and_catch_up() {
3813        let temp_dir = TempDir::new().unwrap();
3814        let address = Address::repeat_byte(0x42);
3815        let chunk = IntegerList::new([100, 200, 300]).unwrap();
3816        let shard_key = ShardedKey::new(address, u64::MAX);
3817
3818        // Write data with a read-write provider
3819        let rw_provider =
3820            RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3821        rw_provider.put::<tables::AccountsHistory>(shard_key, &chunk).unwrap();
3822
3823        // Open read-only provider — it sees the initial data.
3824        let ro_provider = RocksDBBuilder::new(temp_dir.path())
3825            .with_default_tables()
3826            .with_read_only(true)
3827            .build()
3828            .unwrap();
3829
3830        let result =
3831            ro_provider.snapshot().account_history_info(address, 200, None, u64::MAX).unwrap();
3832        assert_eq!(result, HistoryInfo::InChangeset(200));
3833
3834        let result =
3835            ro_provider.snapshot().account_history_info(address, 50, None, u64::MAX).unwrap();
3836        assert_eq!(result, HistoryInfo::NotYetWritten);
3837
3838        let result =
3839            ro_provider.snapshot().account_history_info(address, 400, None, u64::MAX).unwrap();
3840        assert_eq!(result, HistoryInfo::InPlainState);
3841
3842        // Write new data via the primary.
3843        let address2 = Address::repeat_byte(0x43);
3844        let chunk2 = IntegerList::new([500, 600]).unwrap();
3845        let shard_key2 = ShardedKey::new(address2, u64::MAX);
3846        rw_provider.put::<tables::AccountsHistory>(shard_key2, &chunk2).unwrap();
3847
3848        // Read-only doesn't see the new data yet.
3849        let result =
3850            ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3851        assert_eq!(result, HistoryInfo::NotYetWritten);
3852
3853        // Catch up — now it sees the new data.
3854        ro_provider.try_catch_up_with_primary().unwrap();
3855
3856        let result =
3857            ro_provider.snapshot().account_history_info(address2, 500, None, u64::MAX).unwrap();
3858        assert_eq!(result, HistoryInfo::InChangeset(500));
3859    }
3860
3861    #[test]
3862    fn test_account_history_info_ignores_blocks_above_visible_tip() {
3863        let temp_dir = TempDir::new().unwrap();
3864        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3865
3866        let address = Address::repeat_byte(0x42);
3867
3868        provider
3869            .put::<tables::AccountsHistory>(
3870                ShardedKey::new(address, 110),
3871                &IntegerList::new([100, 110]).unwrap(),
3872            )
3873            .unwrap();
3874        provider
3875            .put::<tables::AccountsHistory>(
3876                ShardedKey::new(address, u64::MAX),
3877                &IntegerList::new([200, 210]).unwrap(),
3878            )
3879            .unwrap();
3880
3881        let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3882        assert_eq!(result, HistoryInfo::InPlainState);
3883    }
3884
3885    #[test]
3886    fn test_account_history_info_mixed_shard_respects_visible_tip() {
3887        let temp_dir = TempDir::new().unwrap();
3888        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3889
3890        let address = Address::repeat_byte(0x42);
3891        provider
3892            .put::<tables::AccountsHistory>(
3893                ShardedKey::new(address, u64::MAX),
3894                &IntegerList::new([100, 150, 300]).unwrap(),
3895            )
3896            .unwrap();
3897
3898        let result = provider.snapshot().account_history_info(address, 120, None, 200).unwrap();
3899        assert_eq!(result, HistoryInfo::InChangeset(150));
3900
3901        let result = provider.snapshot().account_history_info(address, 201, None, 200).unwrap();
3902        assert_eq!(result, HistoryInfo::InPlainState);
3903    }
3904
3905    #[test]
3906    fn test_account_history_info_only_stale_entries_use_fallback() {
3907        let temp_dir = TempDir::new().unwrap();
3908        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3909
3910        let address = Address::repeat_byte(0x42);
3911        provider
3912            .put::<tables::AccountsHistory>(
3913                ShardedKey::new(address, u64::MAX),
3914                &IntegerList::new([200, 210]).unwrap(),
3915            )
3916            .unwrap();
3917
3918        let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap();
3919        assert_eq!(result, HistoryInfo::NotYetWritten);
3920
3921        let result =
3922            provider.snapshot().account_history_info(address, 150, Some(100), 150).unwrap();
3923        assert_eq!(result, HistoryInfo::MaybeInPlainState);
3924    }
3925
3926    #[test]
3927    fn test_account_history_shard_split_at_boundary() {
3928        let temp_dir = TempDir::new().unwrap();
3929        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3930
3931        let address = Address::repeat_byte(0x42);
3932        let limit = NUM_OF_INDICES_IN_SHARD;
3933
3934        // Add exactly NUM_OF_INDICES_IN_SHARD + 1 indices to trigger a split
3935        let indices: Vec<u64> = (0..=(limit as u64)).collect();
3936        let mut batch = provider.batch();
3937        batch.append_account_history_shard(address, indices).unwrap();
3938        batch.commit().unwrap();
3939
3940        // Should have 2 shards: one completed shard and one sentinel shard
3941        let completed_key = ShardedKey::new(address, (limit - 1) as u64);
3942        let sentinel_key = ShardedKey::new(address, u64::MAX);
3943
3944        let completed_shard = provider.get::<tables::AccountsHistory>(completed_key).unwrap();
3945        let sentinel_shard = provider.get::<tables::AccountsHistory>(sentinel_key).unwrap();
3946
3947        assert!(completed_shard.is_some(), "completed shard should exist");
3948        assert!(sentinel_shard.is_some(), "sentinel shard should exist");
3949
3950        let completed_shard = completed_shard.unwrap();
3951        let sentinel_shard = sentinel_shard.unwrap();
3952
3953        assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
3954        assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
3955    }
3956
3957    #[test]
3958    fn test_account_history_multiple_shard_splits() {
3959        let temp_dir = TempDir::new().unwrap();
3960        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
3961
3962        let address = Address::repeat_byte(0x43);
3963        let limit = NUM_OF_INDICES_IN_SHARD;
3964
3965        // First batch: add NUM_OF_INDICES_IN_SHARD indices
3966        let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
3967        let mut batch = provider.batch();
3968        batch.append_account_history_shard(address, first_batch_indices).unwrap();
3969        batch.commit().unwrap();
3970
3971        // Should have just a sentinel shard (exactly at limit, not over)
3972        let sentinel_key = ShardedKey::new(address, u64::MAX);
3973        let shard = provider.get::<tables::AccountsHistory>(sentinel_key.clone()).unwrap();
3974        assert!(shard.is_some());
3975        assert_eq!(shard.unwrap().len(), limit as u64);
3976
3977        // Second batch: add another NUM_OF_INDICES_IN_SHARD + 1 indices (causing 2 more shards)
3978        let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
3979        let mut batch = provider.batch();
3980        batch.append_account_history_shard(address, second_batch_indices).unwrap();
3981        batch.commit().unwrap();
3982
3983        // Now we should have: 2 completed shards + 1 sentinel shard
3984        let first_completed = ShardedKey::new(address, (limit - 1) as u64);
3985        let second_completed = ShardedKey::new(address, (2 * limit - 1) as u64);
3986
3987        assert!(
3988            provider.get::<tables::AccountsHistory>(first_completed).unwrap().is_some(),
3989            "first completed shard should exist"
3990        );
3991        assert!(
3992            provider.get::<tables::AccountsHistory>(second_completed).unwrap().is_some(),
3993            "second completed shard should exist"
3994        );
3995        assert!(
3996            provider.get::<tables::AccountsHistory>(sentinel_key).unwrap().is_some(),
3997            "sentinel shard should exist"
3998        );
3999    }
4000
4001    #[test]
4002    fn test_storage_history_shard_split_at_boundary() {
4003        let temp_dir = TempDir::new().unwrap();
4004        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4005
4006        let address = Address::repeat_byte(0x44);
4007        let slot = B256::repeat_byte(0x55);
4008        let limit = NUM_OF_INDICES_IN_SHARD;
4009
4010        // Add exactly NUM_OF_INDICES_IN_SHARD + 1 indices to trigger a split
4011        let indices: Vec<u64> = (0..=(limit as u64)).collect();
4012        let mut batch = provider.batch();
4013        batch.append_storage_history_shard(address, slot, indices).unwrap();
4014        batch.commit().unwrap();
4015
4016        // Should have 2 shards: one completed shard and one sentinel shard
4017        let completed_key = StorageShardedKey::new(address, slot, (limit - 1) as u64);
4018        let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
4019
4020        let completed_shard = provider.get::<tables::StoragesHistory>(completed_key).unwrap();
4021        let sentinel_shard = provider.get::<tables::StoragesHistory>(sentinel_key).unwrap();
4022
4023        assert!(completed_shard.is_some(), "completed shard should exist");
4024        assert!(sentinel_shard.is_some(), "sentinel shard should exist");
4025
4026        let completed_shard = completed_shard.unwrap();
4027        let sentinel_shard = sentinel_shard.unwrap();
4028
4029        assert_eq!(completed_shard.len(), limit as u64, "completed shard should be full");
4030        assert_eq!(sentinel_shard.len(), 1, "sentinel shard should have 1 element");
4031    }
4032
4033    #[test]
4034    fn test_storage_history_multiple_shard_splits() {
4035        let temp_dir = TempDir::new().unwrap();
4036        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4037
4038        let address = Address::repeat_byte(0x46);
4039        let slot = B256::repeat_byte(0x57);
4040        let limit = NUM_OF_INDICES_IN_SHARD;
4041
4042        // First batch: add NUM_OF_INDICES_IN_SHARD indices
4043        let first_batch_indices: Vec<u64> = (0..limit as u64).collect();
4044        let mut batch = provider.batch();
4045        batch.append_storage_history_shard(address, slot, first_batch_indices).unwrap();
4046        batch.commit().unwrap();
4047
4048        // Should have just a sentinel shard (exactly at limit, not over)
4049        let sentinel_key = StorageShardedKey::new(address, slot, u64::MAX);
4050        let shard = provider.get::<tables::StoragesHistory>(sentinel_key.clone()).unwrap();
4051        assert!(shard.is_some());
4052        assert_eq!(shard.unwrap().len(), limit as u64);
4053
4054        // Second batch: add another NUM_OF_INDICES_IN_SHARD + 1 indices (causing 2 more shards)
4055        let second_batch_indices: Vec<u64> = (limit as u64..=(2 * limit) as u64).collect();
4056        let mut batch = provider.batch();
4057        batch.append_storage_history_shard(address, slot, second_batch_indices).unwrap();
4058        batch.commit().unwrap();
4059
4060        // Now we should have: 2 completed shards + 1 sentinel shard
4061        let first_completed = StorageShardedKey::new(address, slot, (limit - 1) as u64);
4062        let second_completed = StorageShardedKey::new(address, slot, (2 * limit - 1) as u64);
4063
4064        assert!(
4065            provider.get::<tables::StoragesHistory>(first_completed).unwrap().is_some(),
4066            "first completed shard should exist"
4067        );
4068        assert!(
4069            provider.get::<tables::StoragesHistory>(second_completed).unwrap().is_some(),
4070            "second completed shard should exist"
4071        );
4072        assert!(
4073            provider.get::<tables::StoragesHistory>(sentinel_key).unwrap().is_some(),
4074            "sentinel shard should exist"
4075        );
4076    }
4077
4078    #[test]
4079    fn test_clear_table() {
4080        let temp_dir = TempDir::new().unwrap();
4081        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4082
4083        let address = Address::repeat_byte(0x42);
4084        let key = ShardedKey::new(address, u64::MAX);
4085        let blocks = BlockNumberList::new_pre_sorted([1, 2, 3]);
4086
4087        provider.put::<tables::AccountsHistory>(key.clone(), &blocks).unwrap();
4088        assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
4089
4090        provider.clear::<tables::AccountsHistory>().unwrap();
4091
4092        assert!(
4093            provider.get::<tables::AccountsHistory>(key).unwrap().is_none(),
4094            "table should be empty after clear"
4095        );
4096        assert!(
4097            provider.first::<tables::AccountsHistory>().unwrap().is_none(),
4098            "first() should return None after clear"
4099        );
4100    }
4101
4102    #[test]
4103    fn test_clear_empty_table() {
4104        let temp_dir = TempDir::new().unwrap();
4105        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4106
4107        assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
4108
4109        provider.clear::<tables::AccountsHistory>().unwrap();
4110
4111        assert!(provider.first::<tables::AccountsHistory>().unwrap().is_none());
4112    }
4113
4114    #[test]
4115    fn test_unwind_account_history_to_basic() {
4116        let temp_dir = TempDir::new().unwrap();
4117        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4118
4119        let address = Address::repeat_byte(0x42);
4120
4121        // Add blocks 0-10
4122        let mut batch = provider.batch();
4123        batch.append_account_history_shard(address, 0..=10).unwrap();
4124        batch.commit().unwrap();
4125
4126        // Verify we have blocks 0-10
4127        let key = ShardedKey::new(address, u64::MAX);
4128        let result = provider.get::<tables::AccountsHistory>(key.clone()).unwrap();
4129        assert!(result.is_some());
4130        let blocks: Vec<u64> = result.unwrap().iter().collect();
4131        assert_eq!(blocks, (0..=10).collect::<Vec<_>>());
4132
4133        // Unwind to block 5 (keep blocks 0-5, remove 6-10)
4134        let mut batch = provider.batch();
4135        batch.unwind_account_history_to(address, 5).unwrap();
4136        batch.commit().unwrap();
4137
4138        // Verify only blocks 0-5 remain
4139        let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4140        assert!(result.is_some());
4141        let blocks: Vec<u64> = result.unwrap().iter().collect();
4142        assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
4143    }
4144
4145    #[test]
4146    fn test_unwind_account_history_to_removes_all() {
4147        let temp_dir = TempDir::new().unwrap();
4148        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4149
4150        let address = Address::repeat_byte(0x42);
4151
4152        // Add blocks 5-10
4153        let mut batch = provider.batch();
4154        batch.append_account_history_shard(address, 5..=10).unwrap();
4155        batch.commit().unwrap();
4156
4157        // Unwind to block 4 (removes all blocks since they're all > 4)
4158        let mut batch = provider.batch();
4159        batch.unwind_account_history_to(address, 4).unwrap();
4160        batch.commit().unwrap();
4161
4162        // Verify no data remains for this address
4163        let key = ShardedKey::new(address, u64::MAX);
4164        let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4165        assert!(result.is_none(), "Should have no data after full unwind");
4166    }
4167
4168    #[test]
4169    fn test_unwind_account_history_to_no_op() {
4170        let temp_dir = TempDir::new().unwrap();
4171        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4172
4173        let address = Address::repeat_byte(0x42);
4174
4175        // Add blocks 0-5
4176        let mut batch = provider.batch();
4177        batch.append_account_history_shard(address, 0..=5).unwrap();
4178        batch.commit().unwrap();
4179
4180        // Unwind to block 10 (no-op since all blocks are <= 10)
4181        let mut batch = provider.batch();
4182        batch.unwind_account_history_to(address, 10).unwrap();
4183        batch.commit().unwrap();
4184
4185        // Verify blocks 0-5 still remain
4186        let key = ShardedKey::new(address, u64::MAX);
4187        let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4188        assert!(result.is_some());
4189        let blocks: Vec<u64> = result.unwrap().iter().collect();
4190        assert_eq!(blocks, (0..=5).collect::<Vec<_>>());
4191    }
4192
4193    #[test]
4194    fn test_unwind_account_history_to_block_zero() {
4195        let temp_dir = TempDir::new().unwrap();
4196        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4197
4198        let address = Address::repeat_byte(0x42);
4199
4200        // Add blocks 0-5 (including block 0)
4201        let mut batch = provider.batch();
4202        batch.append_account_history_shard(address, 0..=5).unwrap();
4203        batch.commit().unwrap();
4204
4205        // Unwind to block 0 (keep only block 0, remove 1-5)
4206        // This simulates the caller doing: unwind_to = min_block.checked_sub(1) where min_block = 1
4207        let mut batch = provider.batch();
4208        batch.unwind_account_history_to(address, 0).unwrap();
4209        batch.commit().unwrap();
4210
4211        // Verify only block 0 remains
4212        let key = ShardedKey::new(address, u64::MAX);
4213        let result = provider.get::<tables::AccountsHistory>(key).unwrap();
4214        assert!(result.is_some());
4215        let blocks: Vec<u64> = result.unwrap().iter().collect();
4216        assert_eq!(blocks, vec![0]);
4217    }
4218
4219    #[test]
4220    fn test_unwind_account_history_to_multi_shard() {
4221        let temp_dir = TempDir::new().unwrap();
4222        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4223
4224        let address = Address::repeat_byte(0x42);
4225
4226        // Create multiple shards by adding more than NUM_OF_INDICES_IN_SHARD entries
4227        // For testing, we'll manually create shards with specific keys
4228        let mut batch = provider.batch();
4229
4230        // First shard: blocks 1-50, keyed by 50
4231        let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4232        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4233
4234        // Second shard: blocks 51-100, keyed by MAX (sentinel)
4235        let shard2 = BlockNumberList::new_pre_sorted(51..=100);
4236        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
4237
4238        batch.commit().unwrap();
4239
4240        // Verify we have 2 shards
4241        let shards = provider.account_history_shards(address).unwrap();
4242        assert_eq!(shards.len(), 2);
4243
4244        // Unwind to block 75 (keep 1-75, remove 76-100)
4245        let mut batch = provider.batch();
4246        batch.unwind_account_history_to(address, 75).unwrap();
4247        batch.commit().unwrap();
4248
4249        // Verify: shard1 should be untouched, shard2 should be truncated
4250        let shards = provider.account_history_shards(address).unwrap();
4251        assert_eq!(shards.len(), 2);
4252
4253        // First shard unchanged
4254        assert_eq!(shards[0].0.highest_block_number, 50);
4255        assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4256
4257        // Second shard truncated and re-keyed to MAX
4258        assert_eq!(shards[1].0.highest_block_number, u64::MAX);
4259        assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
4260    }
4261
4262    #[test]
4263    fn test_unwind_account_history_to_multi_shard_boundary_empty() {
4264        let temp_dir = TempDir::new().unwrap();
4265        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4266
4267        let address = Address::repeat_byte(0x42);
4268
4269        // Create two shards
4270        let mut batch = provider.batch();
4271
4272        // First shard: blocks 1-50, keyed by 50
4273        let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4274        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4275
4276        // Second shard: blocks 75-100, keyed by MAX
4277        let shard2 = BlockNumberList::new_pre_sorted(75..=100);
4278        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard2).unwrap();
4279
4280        batch.commit().unwrap();
4281
4282        // Unwind to block 60 (removes all of shard2 since 75 > 60, promotes shard1 to MAX)
4283        let mut batch = provider.batch();
4284        batch.unwind_account_history_to(address, 60).unwrap();
4285        batch.commit().unwrap();
4286
4287        // Verify: only shard1 remains, now keyed as MAX
4288        let shards = provider.account_history_shards(address).unwrap();
4289        assert_eq!(shards.len(), 1);
4290        assert_eq!(shards[0].0.highest_block_number, u64::MAX);
4291        assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4292    }
4293
4294    #[test]
4295    fn test_account_history_shards_iterator() {
4296        let temp_dir = TempDir::new().unwrap();
4297        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4298
4299        let address = Address::repeat_byte(0x42);
4300        let other_address = Address::repeat_byte(0x43);
4301
4302        // Add data for two addresses
4303        let mut batch = provider.batch();
4304        batch.append_account_history_shard(address, 0..=5).unwrap();
4305        batch.append_account_history_shard(other_address, 10..=15).unwrap();
4306        batch.commit().unwrap();
4307
4308        // Query shards for first address only
4309        let shards = provider.account_history_shards(address).unwrap();
4310        assert_eq!(shards.len(), 1);
4311        assert_eq!(shards[0].0.key, address);
4312
4313        // Query shards for second address only
4314        let shards = provider.account_history_shards(other_address).unwrap();
4315        assert_eq!(shards.len(), 1);
4316        assert_eq!(shards[0].0.key, other_address);
4317
4318        // Query shards for non-existent address
4319        let non_existent = Address::repeat_byte(0x99);
4320        let shards = provider.account_history_shards(non_existent).unwrap();
4321        assert!(shards.is_empty());
4322    }
4323
4324    #[test]
4325    fn test_clear_account_history() {
4326        let temp_dir = TempDir::new().unwrap();
4327        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4328
4329        let address = Address::repeat_byte(0x42);
4330
4331        // Add blocks 0-10
4332        let mut batch = provider.batch();
4333        batch.append_account_history_shard(address, 0..=10).unwrap();
4334        batch.commit().unwrap();
4335
4336        // Clear all history (simulates unwind from block 0)
4337        let mut batch = provider.batch();
4338        batch.clear_account_history(address).unwrap();
4339        batch.commit().unwrap();
4340
4341        // Verify no data remains
4342        let shards = provider.account_history_shards(address).unwrap();
4343        assert!(shards.is_empty(), "All shards should be deleted");
4344    }
4345
4346    #[test]
4347    fn test_unwind_non_sentinel_boundary() {
4348        let temp_dir = TempDir::new().unwrap();
4349        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4350
4351        let address = Address::repeat_byte(0x42);
4352
4353        // Create three shards with non-sentinel boundary
4354        let mut batch = provider.batch();
4355
4356        // Shard 1: blocks 1-50, keyed by 50
4357        let shard1 = BlockNumberList::new_pre_sorted(1..=50);
4358        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 50), &shard1).unwrap();
4359
4360        // Shard 2: blocks 51-100, keyed by 100 (non-sentinel, will be boundary)
4361        let shard2 = BlockNumberList::new_pre_sorted(51..=100);
4362        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, 100), &shard2).unwrap();
4363
4364        // Shard 3: blocks 101-150, keyed by MAX (will be deleted)
4365        let shard3 = BlockNumberList::new_pre_sorted(101..=150);
4366        batch.put::<tables::AccountsHistory>(ShardedKey::new(address, u64::MAX), &shard3).unwrap();
4367
4368        batch.commit().unwrap();
4369
4370        // Verify 3 shards
4371        let shards = provider.account_history_shards(address).unwrap();
4372        assert_eq!(shards.len(), 3);
4373
4374        // Unwind to block 75 (truncates shard2, deletes shard3)
4375        let mut batch = provider.batch();
4376        batch.unwind_account_history_to(address, 75).unwrap();
4377        batch.commit().unwrap();
4378
4379        // Verify: shard1 unchanged, shard2 truncated and re-keyed to MAX, shard3 deleted
4380        let shards = provider.account_history_shards(address).unwrap();
4381        assert_eq!(shards.len(), 2);
4382
4383        // First shard unchanged
4384        assert_eq!(shards[0].0.highest_block_number, 50);
4385        assert_eq!(shards[0].1.iter().collect::<Vec<_>>(), (1..=50).collect::<Vec<_>>());
4386
4387        // Second shard truncated and re-keyed to MAX
4388        assert_eq!(shards[1].0.highest_block_number, u64::MAX);
4389        assert_eq!(shards[1].1.iter().collect::<Vec<_>>(), (51..=75).collect::<Vec<_>>());
4390    }
4391
4392    #[test]
4393    fn test_batch_auto_commit_on_threshold() {
4394        let temp_dir = TempDir::new().unwrap();
4395        let provider =
4396            RocksDBBuilder::new(temp_dir.path()).with_table::<TestTable>().build().unwrap();
4397
4398        // Create batch with tiny threshold (1KB) to force auto-commits
4399        let mut batch = RocksDBBatch {
4400            provider: &provider,
4401            inner: WriteBatchWithTransaction::<true>::default(),
4402            buf: Vec::new(),
4403            auto_commit_threshold: Some(1024), // 1KB
4404        };
4405
4406        // Write entries until we exceed threshold multiple times
4407        // Each entry is ~20 bytes, so 100 entries = ~2KB = 2 auto-commits
4408        for i in 0..100u64 {
4409            let value = format!("value_{i:04}").into_bytes();
4410            batch.put::<TestTable>(i, &value).unwrap();
4411        }
4412
4413        // Data should already be visible (auto-committed) even before final commit
4414        // At least some entries should be readable
4415        let first_visible = provider.get::<TestTable>(0).unwrap();
4416        assert!(first_visible.is_some(), "Auto-committed data should be visible");
4417
4418        // Final commit for remaining batch
4419        batch.commit().unwrap();
4420
4421        // All entries should now be visible
4422        for i in 0..100u64 {
4423            let value = format!("value_{i:04}").into_bytes();
4424            assert_eq!(provider.get::<TestTable>(i).unwrap(), Some(value));
4425        }
4426    }
4427
4428    // ==================== PARAMETERIZED PRUNE TESTS ====================
4429
4430    /// Test case for account history pruning
4431    struct AccountPruneCase {
4432        name: &'static str,
4433        initial_shards: &'static [(u64, &'static [u64])],
4434        prune_to: u64,
4435        expected_outcome: PruneShardOutcome,
4436        expected_shards: &'static [(u64, &'static [u64])],
4437    }
4438
4439    /// Test case for storage history pruning
4440    struct StoragePruneCase {
4441        name: &'static str,
4442        initial_shards: &'static [(u64, &'static [u64])],
4443        prune_to: u64,
4444        expected_outcome: PruneShardOutcome,
4445        expected_shards: &'static [(u64, &'static [u64])],
4446    }
4447
4448    #[test]
4449    fn test_prune_account_history_cases() {
4450        const MAX: u64 = u64::MAX;
4451        const CASES: &[AccountPruneCase] = &[
4452            AccountPruneCase {
4453                name: "single_shard_truncate",
4454                initial_shards: &[(MAX, &[10, 20, 30, 40])],
4455                prune_to: 25,
4456                expected_outcome: PruneShardOutcome::Updated,
4457                expected_shards: &[(MAX, &[30, 40])],
4458            },
4459            AccountPruneCase {
4460                name: "single_shard_delete_all",
4461                initial_shards: &[(MAX, &[10, 20])],
4462                prune_to: 20,
4463                expected_outcome: PruneShardOutcome::Deleted,
4464                expected_shards: &[],
4465            },
4466            AccountPruneCase {
4467                name: "single_shard_noop",
4468                initial_shards: &[(MAX, &[10, 20])],
4469                prune_to: 5,
4470                expected_outcome: PruneShardOutcome::Unchanged,
4471                expected_shards: &[(MAX, &[10, 20])],
4472            },
4473            AccountPruneCase {
4474                name: "no_shards",
4475                initial_shards: &[],
4476                prune_to: 100,
4477                expected_outcome: PruneShardOutcome::Unchanged,
4478                expected_shards: &[],
4479            },
4480            AccountPruneCase {
4481                name: "multi_shard_truncate_first",
4482                initial_shards: &[(30, &[10, 20, 30]), (MAX, &[40, 50, 60])],
4483                prune_to: 25,
4484                expected_outcome: PruneShardOutcome::Updated,
4485                expected_shards: &[(30, &[30]), (MAX, &[40, 50, 60])],
4486            },
4487            AccountPruneCase {
4488                name: "delete_first_shard_sentinel_unchanged",
4489                initial_shards: &[(20, &[10, 20]), (MAX, &[30, 40])],
4490                prune_to: 20,
4491                expected_outcome: PruneShardOutcome::Deleted,
4492                expected_shards: &[(MAX, &[30, 40])],
4493            },
4494            AccountPruneCase {
4495                name: "multi_shard_delete_all_but_last",
4496                initial_shards: &[(10, &[5, 10]), (20, &[15, 20]), (MAX, &[25, 30])],
4497                prune_to: 22,
4498                expected_outcome: PruneShardOutcome::Deleted,
4499                expected_shards: &[(MAX, &[25, 30])],
4500            },
4501            AccountPruneCase {
4502                name: "mid_shard_preserves_key",
4503                initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4504                prune_to: 25,
4505                expected_outcome: PruneShardOutcome::Updated,
4506                expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4507            },
4508            // Equivalence tests
4509            AccountPruneCase {
4510                name: "equiv_delete_early_shards_keep_sentinel",
4511                initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4512                prune_to: 55,
4513                expected_outcome: PruneShardOutcome::Deleted,
4514                expected_shards: &[(MAX, &[60, 70])],
4515            },
4516            AccountPruneCase {
4517                name: "equiv_sentinel_becomes_empty_with_prev",
4518                initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4519                prune_to: 40,
4520                expected_outcome: PruneShardOutcome::Deleted,
4521                expected_shards: &[(MAX, &[50])],
4522            },
4523            AccountPruneCase {
4524                name: "equiv_all_shards_become_empty",
4525                initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4526                prune_to: 51,
4527                expected_outcome: PruneShardOutcome::Deleted,
4528                expected_shards: &[],
4529            },
4530            AccountPruneCase {
4531                name: "equiv_non_sentinel_last_shard_promoted",
4532                initial_shards: &[(100, &[50, 75, 100])],
4533                prune_to: 60,
4534                expected_outcome: PruneShardOutcome::Updated,
4535                expected_shards: &[(MAX, &[75, 100])],
4536            },
4537            AccountPruneCase {
4538                name: "equiv_filter_within_shard",
4539                initial_shards: &[(MAX, &[10, 20, 30, 40])],
4540                prune_to: 25,
4541                expected_outcome: PruneShardOutcome::Updated,
4542                expected_shards: &[(MAX, &[30, 40])],
4543            },
4544            AccountPruneCase {
4545                name: "equiv_multi_shard_partial_delete",
4546                initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4547                prune_to: 35,
4548                expected_outcome: PruneShardOutcome::Deleted,
4549                expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4550            },
4551        ];
4552
4553        let address = Address::repeat_byte(0x42);
4554
4555        for case in CASES {
4556            let temp_dir = TempDir::new().unwrap();
4557            let provider =
4558                RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4559
4560            // Setup initial shards
4561            let mut batch = provider.batch();
4562            for (highest, blocks) in case.initial_shards {
4563                let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4564                batch
4565                    .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4566                    .unwrap();
4567            }
4568            batch.commit().unwrap();
4569
4570            // Prune
4571            let mut batch = provider.batch();
4572            let outcome = batch.prune_account_history_to(address, case.prune_to).unwrap();
4573            batch.commit().unwrap();
4574
4575            // Assert outcome
4576            assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4577
4578            // Assert final shards
4579            let shards = provider.account_history_shards(address).unwrap();
4580            assert_eq!(
4581                shards.len(),
4582                case.expected_shards.len(),
4583                "case '{}': wrong shard count",
4584                case.name
4585            );
4586            for (i, ((key, blocks), (exp_key, exp_blocks))) in
4587                shards.iter().zip(case.expected_shards.iter()).enumerate()
4588            {
4589                assert_eq!(
4590                    key.highest_block_number, *exp_key,
4591                    "case '{}': shard {} wrong key",
4592                    case.name, i
4593                );
4594                assert_eq!(
4595                    blocks.iter().collect::<Vec<_>>(),
4596                    *exp_blocks,
4597                    "case '{}': shard {} wrong blocks",
4598                    case.name,
4599                    i
4600                );
4601            }
4602        }
4603    }
4604
4605    #[test]
4606    fn test_prune_storage_history_cases() {
4607        const MAX: u64 = u64::MAX;
4608        const CASES: &[StoragePruneCase] = &[
4609            StoragePruneCase {
4610                name: "single_shard_truncate",
4611                initial_shards: &[(MAX, &[10, 20, 30, 40])],
4612                prune_to: 25,
4613                expected_outcome: PruneShardOutcome::Updated,
4614                expected_shards: &[(MAX, &[30, 40])],
4615            },
4616            StoragePruneCase {
4617                name: "single_shard_delete_all",
4618                initial_shards: &[(MAX, &[10, 20])],
4619                prune_to: 20,
4620                expected_outcome: PruneShardOutcome::Deleted,
4621                expected_shards: &[],
4622            },
4623            StoragePruneCase {
4624                name: "noop",
4625                initial_shards: &[(MAX, &[10, 20])],
4626                prune_to: 5,
4627                expected_outcome: PruneShardOutcome::Unchanged,
4628                expected_shards: &[(MAX, &[10, 20])],
4629            },
4630            StoragePruneCase {
4631                name: "no_shards",
4632                initial_shards: &[],
4633                prune_to: 100,
4634                expected_outcome: PruneShardOutcome::Unchanged,
4635                expected_shards: &[],
4636            },
4637            StoragePruneCase {
4638                name: "mid_shard_preserves_key",
4639                initial_shards: &[(50, &[10, 20, 30, 40, 50]), (MAX, &[60, 70])],
4640                prune_to: 25,
4641                expected_outcome: PruneShardOutcome::Updated,
4642                expected_shards: &[(50, &[30, 40, 50]), (MAX, &[60, 70])],
4643            },
4644            // Equivalence tests
4645            StoragePruneCase {
4646                name: "equiv_sentinel_promotion",
4647                initial_shards: &[(100, &[50, 75, 100])],
4648                prune_to: 60,
4649                expected_outcome: PruneShardOutcome::Updated,
4650                expected_shards: &[(MAX, &[75, 100])],
4651            },
4652            StoragePruneCase {
4653                name: "equiv_delete_early_shards_keep_sentinel",
4654                initial_shards: &[(20, &[10, 15, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4655                prune_to: 55,
4656                expected_outcome: PruneShardOutcome::Deleted,
4657                expected_shards: &[(MAX, &[60, 70])],
4658            },
4659            StoragePruneCase {
4660                name: "equiv_sentinel_becomes_empty_with_prev",
4661                initial_shards: &[(50, &[30, 40, 50]), (MAX, &[35])],
4662                prune_to: 40,
4663                expected_outcome: PruneShardOutcome::Deleted,
4664                expected_shards: &[(MAX, &[50])],
4665            },
4666            StoragePruneCase {
4667                name: "equiv_all_shards_become_empty",
4668                initial_shards: &[(50, &[30, 40, 50]), (MAX, &[51])],
4669                prune_to: 51,
4670                expected_outcome: PruneShardOutcome::Deleted,
4671                expected_shards: &[],
4672            },
4673            StoragePruneCase {
4674                name: "equiv_filter_within_shard",
4675                initial_shards: &[(MAX, &[10, 20, 30, 40])],
4676                prune_to: 25,
4677                expected_outcome: PruneShardOutcome::Updated,
4678                expected_shards: &[(MAX, &[30, 40])],
4679            },
4680            StoragePruneCase {
4681                name: "equiv_multi_shard_partial_delete",
4682                initial_shards: &[(20, &[10, 20]), (50, &[30, 40, 50]), (MAX, &[60, 70])],
4683                prune_to: 35,
4684                expected_outcome: PruneShardOutcome::Deleted,
4685                expected_shards: &[(50, &[40, 50]), (MAX, &[60, 70])],
4686            },
4687        ];
4688
4689        let address = Address::repeat_byte(0x42);
4690        let storage_key = B256::repeat_byte(0x01);
4691
4692        for case in CASES {
4693            let temp_dir = TempDir::new().unwrap();
4694            let provider =
4695                RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4696
4697            // Setup initial shards
4698            let mut batch = provider.batch();
4699            for (highest, blocks) in case.initial_shards {
4700                let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4701                let key = if *highest == MAX {
4702                    StorageShardedKey::last(address, storage_key)
4703                } else {
4704                    StorageShardedKey::new(address, storage_key, *highest)
4705                };
4706                batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4707            }
4708            batch.commit().unwrap();
4709
4710            // Prune
4711            let mut batch = provider.batch();
4712            let outcome =
4713                batch.prune_storage_history_to(address, storage_key, case.prune_to).unwrap();
4714            batch.commit().unwrap();
4715
4716            // Assert outcome
4717            assert_eq!(outcome, case.expected_outcome, "case '{}': wrong outcome", case.name);
4718
4719            // Assert final shards
4720            let shards = provider.storage_history_shards(address, storage_key).unwrap();
4721            assert_eq!(
4722                shards.len(),
4723                case.expected_shards.len(),
4724                "case '{}': wrong shard count",
4725                case.name
4726            );
4727            for (i, ((key, blocks), (exp_key, exp_blocks))) in
4728                shards.iter().zip(case.expected_shards.iter()).enumerate()
4729            {
4730                assert_eq!(
4731                    key.sharded_key.highest_block_number, *exp_key,
4732                    "case '{}': shard {} wrong key",
4733                    case.name, i
4734                );
4735                assert_eq!(
4736                    blocks.iter().collect::<Vec<_>>(),
4737                    *exp_blocks,
4738                    "case '{}': shard {} wrong blocks",
4739                    case.name,
4740                    i
4741                );
4742            }
4743        }
4744    }
4745
4746    #[test]
4747    fn test_prune_storage_history_does_not_affect_other_slots() {
4748        let temp_dir = TempDir::new().unwrap();
4749        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4750
4751        let address = Address::repeat_byte(0x42);
4752        let slot1 = B256::repeat_byte(0x01);
4753        let slot2 = B256::repeat_byte(0x02);
4754
4755        // Two different storage slots
4756        let mut batch = provider.batch();
4757        batch
4758            .put::<tables::StoragesHistory>(
4759                StorageShardedKey::last(address, slot1),
4760                &BlockNumberList::new_pre_sorted([10u64, 20]),
4761            )
4762            .unwrap();
4763        batch
4764            .put::<tables::StoragesHistory>(
4765                StorageShardedKey::last(address, slot2),
4766                &BlockNumberList::new_pre_sorted([30u64, 40]),
4767            )
4768            .unwrap();
4769        batch.commit().unwrap();
4770
4771        // Prune slot1 to block 20 (deletes all)
4772        let mut batch = provider.batch();
4773        let outcome = batch.prune_storage_history_to(address, slot1, 20).unwrap();
4774        batch.commit().unwrap();
4775
4776        assert_eq!(outcome, PruneShardOutcome::Deleted);
4777
4778        // slot1 should be empty
4779        let shards1 = provider.storage_history_shards(address, slot1).unwrap();
4780        assert!(shards1.is_empty());
4781
4782        // slot2 should be unchanged
4783        let shards2 = provider.storage_history_shards(address, slot2).unwrap();
4784        assert_eq!(shards2.len(), 1);
4785        assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![30, 40]);
4786    }
4787
4788    #[test]
4789    fn test_prune_invariants() {
4790        // Test invariants: no empty shards, sentinel is always last
4791        let address = Address::repeat_byte(0x42);
4792        let storage_key = B256::repeat_byte(0x01);
4793
4794        // Test cases that exercise invariants
4795        #[expect(clippy::type_complexity)]
4796        let invariant_cases: &[(&[(u64, &[u64])], u64)] = &[
4797            // Account: shards where middle becomes empty
4798            (&[(10, &[5, 10]), (20, &[15, 20]), (u64::MAX, &[25, 30])], 20),
4799            // Account: non-sentinel shard only, partial prune -> must become sentinel
4800            (&[(100, &[50, 100])], 60),
4801        ];
4802
4803        for (initial_shards, prune_to) in invariant_cases {
4804            // Test account history invariants
4805            {
4806                let temp_dir = TempDir::new().unwrap();
4807                let provider =
4808                    RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4809
4810                let mut batch = provider.batch();
4811                for (highest, blocks) in *initial_shards {
4812                    let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4813                    batch
4814                        .put::<tables::AccountsHistory>(ShardedKey::new(address, *highest), &shard)
4815                        .unwrap();
4816                }
4817                batch.commit().unwrap();
4818
4819                let mut batch = provider.batch();
4820                batch.prune_account_history_to(address, *prune_to).unwrap();
4821                batch.commit().unwrap();
4822
4823                let shards = provider.account_history_shards(address).unwrap();
4824
4825                // Invariant 1: no empty shards
4826                for (key, blocks) in &shards {
4827                    assert!(
4828                        !blocks.is_empty(),
4829                        "Account: empty shard at key {}",
4830                        key.highest_block_number
4831                    );
4832                }
4833
4834                // Invariant 2: last shard is sentinel
4835                if !shards.is_empty() {
4836                    let last = shards.last().unwrap();
4837                    assert_eq!(
4838                        last.0.highest_block_number,
4839                        u64::MAX,
4840                        "Account: last shard must be sentinel"
4841                    );
4842                }
4843            }
4844
4845            // Test storage history invariants
4846            {
4847                let temp_dir = TempDir::new().unwrap();
4848                let provider =
4849                    RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4850
4851                let mut batch = provider.batch();
4852                for (highest, blocks) in *initial_shards {
4853                    let shard = BlockNumberList::new_pre_sorted(blocks.iter().copied());
4854                    let key = if *highest == u64::MAX {
4855                        StorageShardedKey::last(address, storage_key)
4856                    } else {
4857                        StorageShardedKey::new(address, storage_key, *highest)
4858                    };
4859                    batch.put::<tables::StoragesHistory>(key, &shard).unwrap();
4860                }
4861                batch.commit().unwrap();
4862
4863                let mut batch = provider.batch();
4864                batch.prune_storage_history_to(address, storage_key, *prune_to).unwrap();
4865                batch.commit().unwrap();
4866
4867                let shards = provider.storage_history_shards(address, storage_key).unwrap();
4868
4869                // Invariant 1: no empty shards
4870                for (key, blocks) in &shards {
4871                    assert!(
4872                        !blocks.is_empty(),
4873                        "Storage: empty shard at key {}",
4874                        key.sharded_key.highest_block_number
4875                    );
4876                }
4877
4878                // Invariant 2: last shard is sentinel
4879                if !shards.is_empty() {
4880                    let last = shards.last().unwrap();
4881                    assert_eq!(
4882                        last.0.sharded_key.highest_block_number,
4883                        u64::MAX,
4884                        "Storage: last shard must be sentinel"
4885                    );
4886                }
4887            }
4888        }
4889    }
4890
4891    #[test]
4892    fn test_prune_account_history_batch_multiple_sorted_targets() {
4893        let temp_dir = TempDir::new().unwrap();
4894        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4895
4896        let addr1 = Address::repeat_byte(0x01);
4897        let addr2 = Address::repeat_byte(0x02);
4898        let addr3 = Address::repeat_byte(0x03);
4899
4900        // Setup shards for each address
4901        let mut batch = provider.batch();
4902        batch
4903            .put::<tables::AccountsHistory>(
4904                ShardedKey::new(addr1, u64::MAX),
4905                &BlockNumberList::new_pre_sorted([10, 20, 30]),
4906            )
4907            .unwrap();
4908        batch
4909            .put::<tables::AccountsHistory>(
4910                ShardedKey::new(addr2, u64::MAX),
4911                &BlockNumberList::new_pre_sorted([5, 10, 15]),
4912            )
4913            .unwrap();
4914        batch
4915            .put::<tables::AccountsHistory>(
4916                ShardedKey::new(addr3, u64::MAX),
4917                &BlockNumberList::new_pre_sorted([100, 200]),
4918            )
4919            .unwrap();
4920        batch.commit().unwrap();
4921
4922        // Prune all three (sorted by address)
4923        let mut targets = vec![(addr1, 15), (addr2, 10), (addr3, 50)];
4924        targets.sort_by_key(|(addr, _)| *addr);
4925
4926        let mut batch = provider.batch();
4927        let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4928        batch.commit().unwrap();
4929
4930        // addr1: prune <=15, keep [20, 30] -> updated
4931        // addr2: prune <=10, keep [15] -> updated
4932        // addr3: prune <=50, keep [100, 200] -> unchanged
4933        assert_eq!(outcomes.updated, 2);
4934        assert_eq!(outcomes.unchanged, 1);
4935
4936        let shards1 = provider.account_history_shards(addr1).unwrap();
4937        assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
4938
4939        let shards2 = provider.account_history_shards(addr2).unwrap();
4940        assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15]);
4941
4942        let shards3 = provider.account_history_shards(addr3).unwrap();
4943        assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![100, 200]);
4944    }
4945
4946    #[test]
4947    fn test_prune_account_history_batch_target_with_no_shards() {
4948        let temp_dir = TempDir::new().unwrap();
4949        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4950
4951        let addr1 = Address::repeat_byte(0x01);
4952        let addr2 = Address::repeat_byte(0x02); // No shards for this one
4953        let addr3 = Address::repeat_byte(0x03);
4954
4955        // Only setup shards for addr1 and addr3
4956        let mut batch = provider.batch();
4957        batch
4958            .put::<tables::AccountsHistory>(
4959                ShardedKey::new(addr1, u64::MAX),
4960                &BlockNumberList::new_pre_sorted([10, 20]),
4961            )
4962            .unwrap();
4963        batch
4964            .put::<tables::AccountsHistory>(
4965                ShardedKey::new(addr3, u64::MAX),
4966                &BlockNumberList::new_pre_sorted([30, 40]),
4967            )
4968            .unwrap();
4969        batch.commit().unwrap();
4970
4971        // Prune all three (addr2 has no shards - tests p > target_prefix case)
4972        let mut targets = vec![(addr1, 15), (addr2, 100), (addr3, 35)];
4973        targets.sort_by_key(|(addr, _)| *addr);
4974
4975        let mut batch = provider.batch();
4976        let outcomes = batch.prune_account_history_batch(&targets).unwrap();
4977        batch.commit().unwrap();
4978
4979        // addr1: updated (keep [20])
4980        // addr2: unchanged (no shards)
4981        // addr3: updated (keep [40])
4982        assert_eq!(outcomes.updated, 2);
4983        assert_eq!(outcomes.unchanged, 1);
4984
4985        let shards1 = provider.account_history_shards(addr1).unwrap();
4986        assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20]);
4987
4988        let shards3 = provider.account_history_shards(addr3).unwrap();
4989        assert_eq!(shards3[0].1.iter().collect::<Vec<_>>(), vec![40]);
4990    }
4991
4992    #[test]
4993    fn test_prune_storage_history_batch_multiple_sorted_targets() {
4994        let temp_dir = TempDir::new().unwrap();
4995        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
4996
4997        let addr = Address::repeat_byte(0x42);
4998        let slot1 = B256::repeat_byte(0x01);
4999        let slot2 = B256::repeat_byte(0x02);
5000
5001        // Setup shards
5002        let mut batch = provider.batch();
5003        batch
5004            .put::<tables::StoragesHistory>(
5005                StorageShardedKey::new(addr, slot1, u64::MAX),
5006                &BlockNumberList::new_pre_sorted([10, 20, 30]),
5007            )
5008            .unwrap();
5009        batch
5010            .put::<tables::StoragesHistory>(
5011                StorageShardedKey::new(addr, slot2, u64::MAX),
5012                &BlockNumberList::new_pre_sorted([5, 15, 25]),
5013            )
5014            .unwrap();
5015        batch.commit().unwrap();
5016
5017        // Prune both (sorted)
5018        let mut targets = vec![((addr, slot1), 15), ((addr, slot2), 10)];
5019        targets.sort_by_key(|((a, s), _)| (*a, *s));
5020
5021        let mut batch = provider.batch();
5022        let outcomes = batch.prune_storage_history_batch(&targets).unwrap();
5023        batch.commit().unwrap();
5024
5025        assert_eq!(outcomes.updated, 2);
5026
5027        let shards1 = provider.storage_history_shards(addr, slot1).unwrap();
5028        assert_eq!(shards1[0].1.iter().collect::<Vec<_>>(), vec![20, 30]);
5029
5030        let shards2 = provider.storage_history_shards(addr, slot2).unwrap();
5031        assert_eq!(shards2[0].1.iter().collect::<Vec<_>>(), vec![15, 25]);
5032    }
5033
5034    /// Shards for one address, keyed by highest block, as `(highest, blocks)`.
5035    fn account_shard_layout(provider: &RocksDBProvider, address: Address) -> Vec<(u64, Vec<u64>)> {
5036        provider
5037            .account_history_shards(address)
5038            .unwrap()
5039            .into_iter()
5040            .map(|(key, list)| (key.highest_block_number, list.iter().collect::<Vec<_>>()))
5041            .collect()
5042    }
5043
5044    /// Shards for one storage slot, keyed by highest block, as `(highest, blocks)`.
5045    fn storage_shard_layout(
5046        provider: &RocksDBProvider,
5047        address: Address,
5048        storage_key: B256,
5049    ) -> Vec<(u64, Vec<u64>)> {
5050        provider
5051            .storage_history_shards(address, storage_key)
5052            .unwrap()
5053            .into_iter()
5054            .map(|(key, list)| {
5055                (key.sharded_key.highest_block_number, list.iter().collect::<Vec<_>>())
5056            })
5057            .collect()
5058    }
5059
5060    fn seed_three_storage_shards(provider: &RocksDBProvider, address: Address, storage_key: B256) {
5061        let mut batch = provider.batch();
5062        batch
5063            .put::<tables::StoragesHistory>(
5064                StorageShardedKey::new(address, storage_key, 100),
5065                &BlockNumberList::new_pre_sorted([10, 50, 100]),
5066            )
5067            .unwrap();
5068        batch
5069            .put::<tables::StoragesHistory>(
5070                StorageShardedKey::new(address, storage_key, 200),
5071                &BlockNumberList::new_pre_sorted([150, 200]),
5072            )
5073            .unwrap();
5074        batch
5075            .put::<tables::StoragesHistory>(
5076                StorageShardedKey::last(address, storage_key),
5077                &BlockNumberList::new_pre_sorted([250, 300]),
5078            )
5079            .unwrap();
5080        batch.commit().unwrap();
5081    }
5082
5083    #[test]
5084    fn test_prune_storage_history_batch_trims_sentinel_once_earlier_shards_expire() {
5085        let temp_dir = TempDir::new().unwrap();
5086        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
5087
5088        let addr = Address::repeat_byte(0x42);
5089        let slot = B256::repeat_byte(0x01);
5090        seed_three_storage_shards(&provider, addr, slot);
5091
5092        // Every non-sentinel shard expires whole and the sentinel loses its lowest block.
5093        let mut batch = provider.batch();
5094        let outcomes = batch.prune_storage_history_batch(&[((addr, slot), 250)]).unwrap();
5095        batch.commit().unwrap();
5096
5097        assert_eq!(outcomes.deleted, 1);
5098        assert_eq!(storage_shard_layout(&provider, addr, slot), vec![(u64::MAX, vec![300])]);
5099    }
5100
5101    #[test]
5102    fn test_prune_account_history_batch_seeks_after_stopping_early() {
5103        let temp_dir = TempDir::new().unwrap();
5104        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
5105
5106        let addr1 = Address::repeat_byte(0x01);
5107        let addr2 = Address::repeat_byte(0x02);
5108
5109        let mut batch = provider.batch();
5110        batch
5111            .put::<tables::AccountsHistory>(
5112                ShardedKey::new(addr1, 100),
5113                &BlockNumberList::new_pre_sorted([10, 50, 100]),
5114            )
5115            .unwrap();
5116        batch
5117            .put::<tables::AccountsHistory>(
5118                ShardedKey::new(addr1, u64::MAX),
5119                &BlockNumberList::new_pre_sorted([250, 300]),
5120            )
5121            .unwrap();
5122        batch
5123            .put::<tables::AccountsHistory>(
5124                ShardedKey::new(addr2, u64::MAX),
5125                &BlockNumberList::new_pre_sorted([5, 10, 15]),
5126            )
5127            .unwrap();
5128        batch.commit().unwrap();
5129
5130        // The first target stops on addr1's oldest shard, leaving the iterator on addr1's
5131        // sentinel. The second target must seek past it instead of skipping addr2.
5132        let mut batch = provider.batch();
5133        let outcomes = batch.prune_account_history_batch(&[(addr1, 50), (addr2, 10)]).unwrap();
5134        batch.commit().unwrap();
5135
5136        assert_eq!(outcomes.updated, 2);
5137        assert_eq!(
5138            account_shard_layout(&provider, addr1),
5139            vec![(100, vec![100]), (u64::MAX, vec![250, 300])]
5140        );
5141        assert_eq!(account_shard_layout(&provider, addr2), vec![(u64::MAX, vec![15])]);
5142    }
5143
5144    #[test]
5145    fn test_prune_storage_history_batch_seeks_after_stopping_early() {
5146        let temp_dir = TempDir::new().unwrap();
5147        let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap();
5148
5149        let addr = Address::repeat_byte(0x42);
5150        let slot1 = B256::repeat_byte(0x01);
5151        let slot2 = B256::repeat_byte(0x02);
5152        seed_three_storage_shards(&provider, addr, slot1);
5153
5154        let mut batch = provider.batch();
5155        batch
5156            .put::<tables::StoragesHistory>(
5157                StorageShardedKey::last(addr, slot2),
5158                &BlockNumberList::new_pre_sorted([20, 40]),
5159            )
5160            .unwrap();
5161        batch.commit().unwrap();
5162
5163        // The first target stops on slot1's oldest shard, leaving the iterator on slot1's next
5164        // shard. The second target must seek past it instead of skipping slot2.
5165        let mut batch = provider.batch();
5166        let outcomes =
5167            batch.prune_storage_history_batch(&[((addr, slot1), 50), ((addr, slot2), 30)]).unwrap();
5168        batch.commit().unwrap();
5169
5170        assert_eq!(outcomes.updated, 2);
5171        assert_eq!(
5172            storage_shard_layout(&provider, addr, slot1),
5173            vec![(100, vec![100]), (200, vec![150, 200]), (u64::MAX, vec![250, 300])]
5174        );
5175        assert_eq!(storage_shard_layout(&provider, addr, slot2), vec![(u64::MAX, vec![40])]);
5176    }
5177}