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