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