Skip to main content

reth_provider/
either_writer.rs

1//! Generic reader and writer abstractions for interacting with either database tables or static
2//! files.
3
4use std::{
5    collections::BTreeSet,
6    marker::PhantomData,
7    ops::{Range, RangeInclusive},
8};
9
10use crate::{
11    providers::{
12        history_info, rocksdb::RocksDBBatch, HistoryInfo, StaticFileProvider,
13        StaticFileProviderRWRefMut,
14    },
15    StaticFileProviderFactory,
16};
17use alloy_primitives::{map::HashMap, Address, BlockNumber, TxHash, TxNumber, B256};
18use rayon::slice::ParallelSliceMut;
19use reth_db::{
20    cursor::{DbCursorRO, DbDupCursorRW},
21    models::{AccountBeforeTx, StorageBeforeTx},
22    static_file::TransactionSenderMask,
23    table::Value,
24    transaction::{CursorMutTy, CursorTy, DbTx, DbTxMut, DupCursorMutTy, DupCursorTy},
25};
26use reth_db_api::{
27    cursor::DbCursorRW,
28    models::{storage_sharded_key::StorageShardedKey, BlockNumberAddress, ShardedKey},
29    tables,
30    tables::BlockNumberList,
31};
32use reth_errors::ProviderError;
33use reth_node_types::NodePrimitives;
34use reth_primitives_traits::{ReceiptTy, StorageEntry};
35use reth_static_file_types::StaticFileSegment;
36use reth_storage_api::{
37    ChangeSetReader, DBProvider, DbTxProvider, NodePrimitivesProvider, StorageSettingsCache,
38};
39use reth_storage_errors::provider::ProviderResult;
40use strum::{Display, EnumIs};
41
42/// Type alias for [`EitherReader`] constructors.
43type EitherReaderTy<'a, P, T> = EitherReader<
44    'a,
45    CursorTy<<P as DbTxProvider>::Tx, T>,
46    <P as NodePrimitivesProvider>::Primitives,
47>;
48
49/// Type alias for [`EitherReader`] constructors.
50type DupEitherReaderTy<'a, P, T> = EitherReader<
51    'a,
52    DupCursorTy<<P as DbTxProvider>::Tx, T>,
53    <P as NodePrimitivesProvider>::Primitives,
54>;
55
56/// Type alias for dup [`EitherWriter`] constructors.
57type DupEitherWriterTy<'a, P, T> = EitherWriter<
58    'a,
59    DupCursorMutTy<<P as DbTxProvider>::Tx, T>,
60    <P as NodePrimitivesProvider>::Primitives,
61>;
62
63/// Type alias for [`EitherWriter`] constructors.
64type EitherWriterTy<'a, P, T> = EitherWriter<
65    'a,
66    CursorMutTy<<P as DbTxProvider>::Tx, T>,
67    <P as NodePrimitivesProvider>::Primitives,
68>;
69
70/// Helper type for `RocksDB` batch argument in writer constructors.
71pub type RocksBatchArg<'a> = crate::providers::rocksdb::RocksDBBatch<'a>;
72
73/// The raw `RocksDB` batch type returned by [`EitherWriter::into_raw_rocksdb_batch`].
74pub type RawRocksDBBatch = rocksdb::WriteBatchWithTransaction<true>;
75
76/// Helper type for `RocksDB` snapshot argument in reader constructors.
77///
78/// The `Option` allows callers to skip `RocksDB` access when it isn't needed
79/// (e.g., on legacy MDBX-only nodes).
80pub type RocksDBRefArg<'a> = Option<crate::providers::rocksdb::RocksReadSnapshot<'a>>;
81
82/// Represents a destination for writing data, either to database, static files, or `RocksDB`.
83#[derive(Debug, Display)]
84pub enum EitherWriter<'a, CURSOR, N> {
85    /// Write to database table via cursor
86    Database(CURSOR),
87    /// Write to static file
88    StaticFile(StaticFileProviderRWRefMut<'a, N>),
89    /// Write to `RocksDB` using a write-only batch (historical tables).
90    RocksDB(RocksDBBatch<'a>),
91}
92
93impl<'a> EitherWriter<'a, (), ()> {
94    /// Creates a new [`EitherWriter`] for receipts based on storage settings and prune modes.
95    pub fn new_receipts<P>(
96        provider: &'a P,
97        block_number: BlockNumber,
98    ) -> ProviderResult<EitherWriterTy<'a, P, tables::Receipts<ReceiptTy<P::Primitives>>>>
99    where
100        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache + StaticFileProviderFactory,
101        P::Tx: DbTxMut,
102        ReceiptTy<P::Primitives>: Value,
103    {
104        if Self::receipts_destination(provider).is_static_file() {
105            Ok(EitherWriter::StaticFile(
106                provider.get_static_file_writer(block_number, StaticFileSegment::Receipts)?,
107            ))
108        } else {
109            Ok(EitherWriter::Database(
110                provider.tx_ref().cursor_write::<tables::Receipts<ReceiptTy<P::Primitives>>>()?,
111            ))
112        }
113    }
114
115    /// Creates a new [`EitherWriter`] for senders based on storage settings.
116    pub fn new_senders<P>(
117        provider: &'a P,
118        block_number: BlockNumber,
119    ) -> ProviderResult<EitherWriterTy<'a, P, tables::TransactionSenders>>
120    where
121        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache + StaticFileProviderFactory,
122        P::Tx: DbTxMut,
123    {
124        if EitherWriterDestination::senders(provider).is_static_file() {
125            Ok(EitherWriter::StaticFile(
126                provider
127                    .get_static_file_writer(block_number, StaticFileSegment::TransactionSenders)?,
128            ))
129        } else {
130            Ok(EitherWriter::Database(
131                provider.tx_ref().cursor_write::<tables::TransactionSenders>()?,
132            ))
133        }
134    }
135
136    /// Creates a new [`EitherWriter`] for account changesets based on storage settings and prune
137    /// modes.
138    pub fn new_account_changesets<P>(
139        provider: &'a P,
140        block_number: BlockNumber,
141    ) -> ProviderResult<DupEitherWriterTy<'a, P, tables::AccountChangeSets>>
142    where
143        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache + StaticFileProviderFactory,
144        P::Tx: DbTxMut,
145    {
146        if provider.cached_storage_settings().storage_v2 {
147            Ok(EitherWriter::StaticFile(
148                provider
149                    .get_static_file_writer(block_number, StaticFileSegment::AccountChangeSets)?,
150            ))
151        } else {
152            Ok(EitherWriter::Database(
153                provider.tx_ref().cursor_dup_write::<tables::AccountChangeSets>()?,
154            ))
155        }
156    }
157
158    /// Creates a new [`EitherWriter`] for storage changesets based on storage settings.
159    pub fn new_storage_changesets<P>(
160        provider: &'a P,
161        block_number: BlockNumber,
162    ) -> ProviderResult<DupEitherWriterTy<'a, P, tables::StorageChangeSets>>
163    where
164        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache + StaticFileProviderFactory,
165        P::Tx: DbTxMut,
166    {
167        if provider.cached_storage_settings().storage_v2 {
168            Ok(EitherWriter::StaticFile(
169                provider
170                    .get_static_file_writer(block_number, StaticFileSegment::StorageChangeSets)?,
171            ))
172        } else {
173            Ok(EitherWriter::Database(
174                provider.tx_ref().cursor_dup_write::<tables::StorageChangeSets>()?,
175            ))
176        }
177    }
178
179    /// Returns the destination for writing receipts.
180    ///
181    /// The rules are as follows:
182    /// - If the node should not always write receipts to static files, and any receipt pruning is
183    ///   enabled, write to the database.
184    /// - If the node should always write receipts to static files, but receipt log filter pruning
185    ///   is enabled, write to the database.
186    /// - Otherwise, write to static files.
187    pub fn receipts_destination<P: DBProvider + StorageSettingsCache>(
188        provider: &P,
189    ) -> EitherWriterDestination {
190        let receipts_in_static_files = provider.cached_storage_settings().storage_v2;
191        let prune_modes = provider.prune_modes_ref();
192
193        if !receipts_in_static_files && prune_modes.has_receipts_pruning() ||
194            // TODO: support writing receipts to static files with log filter pruning enabled
195            receipts_in_static_files && !prune_modes.receipts_log_filter.is_empty()
196        {
197            EitherWriterDestination::Database
198        } else {
199            EitherWriterDestination::StaticFile
200        }
201    }
202
203    /// Returns the destination for writing account changesets.
204    ///
205    /// This determines the destination based solely on storage settings.
206    pub fn account_changesets_destination<P: DBProvider + StorageSettingsCache>(
207        provider: &P,
208    ) -> EitherWriterDestination {
209        if provider.cached_storage_settings().storage_v2 {
210            EitherWriterDestination::StaticFile
211        } else {
212            EitherWriterDestination::Database
213        }
214    }
215
216    /// Returns the destination for writing storage changesets.
217    ///
218    /// This determines the destination based solely on storage settings.
219    pub fn storage_changesets_destination<P: DBProvider + StorageSettingsCache>(
220        provider: &P,
221    ) -> EitherWriterDestination {
222        if provider.cached_storage_settings().storage_v2 {
223            EitherWriterDestination::StaticFile
224        } else {
225            EitherWriterDestination::Database
226        }
227    }
228
229    /// Creates a new [`EitherWriter`] for storages history based on storage settings.
230    pub fn new_storages_history<P>(
231        provider: &P,
232        _rocksdb_batch: RocksBatchArg<'a>,
233    ) -> ProviderResult<EitherWriterTy<'a, P, tables::StoragesHistory>>
234    where
235        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
236        P::Tx: DbTxMut,
237    {
238        if provider.cached_storage_settings().storage_v2 {
239            return Ok(EitherWriter::RocksDB(_rocksdb_batch));
240        }
241
242        Ok(EitherWriter::Database(provider.tx_ref().cursor_write::<tables::StoragesHistory>()?))
243    }
244
245    /// Creates a new [`EitherWriter`] for transaction hash numbers based on storage settings.
246    pub fn new_transaction_hash_numbers<P>(
247        provider: &P,
248        _rocksdb_batch: RocksBatchArg<'a>,
249    ) -> ProviderResult<EitherWriterTy<'a, P, tables::TransactionHashNumbers>>
250    where
251        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
252        P::Tx: DbTxMut,
253    {
254        if provider.cached_storage_settings().storage_v2 {
255            return Ok(EitherWriter::RocksDB(_rocksdb_batch));
256        }
257
258        Ok(EitherWriter::Database(
259            provider.tx_ref().cursor_write::<tables::TransactionHashNumbers>()?,
260        ))
261    }
262
263    /// Creates a new [`EitherWriter`] for account history based on storage settings.
264    pub fn new_accounts_history<P>(
265        provider: &P,
266        _rocksdb_batch: RocksBatchArg<'a>,
267    ) -> ProviderResult<EitherWriterTy<'a, P, tables::AccountsHistory>>
268    where
269        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
270        P::Tx: DbTxMut,
271    {
272        if provider.cached_storage_settings().storage_v2 {
273            return Ok(EitherWriter::RocksDB(_rocksdb_batch));
274        }
275
276        Ok(EitherWriter::Database(provider.tx_ref().cursor_write::<tables::AccountsHistory>()?))
277    }
278}
279
280impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N> {
281    /// Extracts the raw `RocksDB` write batch from this writer, if it contains one.
282    ///
283    /// Returns `Some(WriteBatchWithTransaction)` for [`Self::RocksDB`] variant,
284    /// `None` for other variants.
285    ///
286    /// This is used to defer `RocksDB` commits to the provider level, ensuring all
287    /// storage commits (MDBX, static files, `RocksDB`) happen atomically in a single place.
288    pub fn into_raw_rocksdb_batch(self) -> Option<rocksdb::WriteBatchWithTransaction<true>> {
289        match self {
290            Self::Database(_) | Self::StaticFile(_) => None,
291            Self::RocksDB(batch) => Some(batch.into_inner()),
292        }
293    }
294
295    /// Increment the block number.
296    ///
297    /// Relevant only for [`Self::StaticFile`]. It is a no-op for [`Self::Database`].
298    pub fn increment_block(&mut self, expected_block_number: BlockNumber) -> ProviderResult<()> {
299        match self {
300            Self::Database(_) => Ok(()),
301            Self::StaticFile(writer) => writer.increment_block(expected_block_number),
302            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
303        }
304    }
305
306    /// Ensures that the writer is positioned at the specified block number.
307    ///
308    /// If the writer is positioned at a greater block number than the specified one, the writer
309    /// will NOT be unwound and the error will be returned.
310    ///
311    /// Relevant only for [`Self::StaticFile`]. It is a no-op for [`Self::Database`].
312    pub fn ensure_at_block(&mut self, block_number: BlockNumber) -> ProviderResult<()> {
313        match self {
314            Self::Database(_) => Ok(()),
315            Self::StaticFile(writer) => writer.ensure_at_block(block_number),
316            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
317        }
318    }
319}
320
321impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
322where
323    N::Receipt: Value,
324    CURSOR: DbCursorRW<tables::Receipts<N::Receipt>>,
325{
326    /// Append a transaction receipt.
327    pub fn append_receipt(&mut self, tx_num: TxNumber, receipt: &N::Receipt) -> ProviderResult<()> {
328        match self {
329            Self::Database(cursor) => Ok(cursor.append(tx_num, receipt)?),
330            Self::StaticFile(writer) => writer.append_receipt(tx_num, receipt),
331            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
332        }
333    }
334}
335
336impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
337where
338    CURSOR: DbCursorRW<tables::TransactionSenders>,
339{
340    /// Append a transaction sender to the destination
341    pub fn append_sender(&mut self, tx_num: TxNumber, sender: &Address) -> ProviderResult<()> {
342        match self {
343            Self::Database(cursor) => Ok(cursor.append(tx_num, sender)?),
344            Self::StaticFile(writer) => writer.append_transaction_sender(tx_num, sender),
345            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
346        }
347    }
348
349    /// Append transaction senders to the destination
350    pub fn append_senders<I>(&mut self, senders: I) -> ProviderResult<()>
351    where
352        I: Iterator<Item = (TxNumber, Address)>,
353    {
354        match self {
355            Self::Database(cursor) => {
356                for (tx_num, sender) in senders {
357                    cursor.append(tx_num, &sender)?;
358                }
359                Ok(())
360            }
361            Self::StaticFile(writer) => writer.append_transaction_senders(senders),
362            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
363        }
364    }
365
366    /// Removes all transaction senders above the given transaction number, and stops at the given
367    /// block number.
368    pub fn prune_senders(
369        &mut self,
370        unwind_tx_from: TxNumber,
371        block: BlockNumber,
372    ) -> ProviderResult<()>
373    where
374        CURSOR: DbCursorRO<tables::TransactionSenders>,
375    {
376        match self {
377            Self::Database(cursor) => {
378                let mut walker = cursor.walk_range(unwind_tx_from..)?;
379                while walker.next().transpose()?.is_some() {
380                    walker.delete_current()?;
381                }
382            }
383            Self::StaticFile(writer) => {
384                let static_file_transaction_sender_num = writer
385                    .reader()
386                    .get_highest_static_file_tx(StaticFileSegment::TransactionSenders);
387
388                let to_delete = static_file_transaction_sender_num
389                    .map(|static_num| (static_num + 1).saturating_sub(unwind_tx_from))
390                    .unwrap_or_default();
391
392                writer.prune_transaction_senders(to_delete, block)?;
393            }
394            Self::RocksDB(_) => return Err(ProviderError::UnsupportedProvider),
395        }
396
397        Ok(())
398    }
399}
400
401impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
402where
403    CURSOR: DbCursorRW<tables::TransactionHashNumbers> + DbCursorRO<tables::TransactionHashNumbers>,
404{
405    /// Puts a transaction hash number mapping.
406    ///
407    /// When `append_only` is true, uses `cursor.append()` which is significantly faster
408    /// but requires entries to be inserted in order and the table to be empty.
409    /// When false, uses `cursor.upsert()` which handles arbitrary insertion order and duplicates.
410    pub fn put_transaction_hash_number(
411        &mut self,
412        hash: TxHash,
413        tx_num: TxNumber,
414        append_only: bool,
415    ) -> ProviderResult<()> {
416        match self {
417            Self::Database(cursor) => {
418                if append_only {
419                    Ok(cursor.append(hash, &tx_num)?)
420                } else {
421                    Ok(cursor.upsert(hash, &tx_num)?)
422                }
423            }
424            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
425            Self::RocksDB(batch) => batch.put::<tables::TransactionHashNumbers>(hash, &tx_num),
426        }
427    }
428
429    /// Puts multiple transaction hash number mappings in a batch.
430    ///
431    /// Accepts a vector of `(TxHash, TxNumber)` tuples and writes them all using the same cursor.
432    /// This is more efficient than calling `put_transaction_hash_number` repeatedly.
433    ///
434    /// When `append_only` is true, uses `cursor.append()` which requires entries to be
435    /// pre-sorted and the table to be empty or have only lower keys.
436    /// When false, uses `cursor.upsert()` which handles arbitrary insertion order.
437    pub fn put_transaction_hash_numbers_batch(
438        &mut self,
439        entries: Vec<(TxHash, TxNumber)>,
440        append_only: bool,
441    ) -> ProviderResult<()> {
442        match self {
443            Self::Database(cursor) => {
444                for (hash, tx_num) in entries {
445                    if append_only {
446                        cursor.append(hash, &tx_num)?;
447                    } else {
448                        cursor.upsert(hash, &tx_num)?;
449                    }
450                }
451                Ok(())
452            }
453            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
454            Self::RocksDB(batch) => {
455                for (hash, tx_num) in entries {
456                    batch.put::<tables::TransactionHashNumbers>(hash, &tx_num)?;
457                }
458                Ok(())
459            }
460        }
461    }
462
463    /// Deletes a transaction hash number mapping.
464    pub fn delete_transaction_hash_number(&mut self, hash: TxHash) -> ProviderResult<()> {
465        match self {
466            Self::Database(cursor) => {
467                if cursor.seek_exact(hash)?.is_some() {
468                    cursor.delete_current()?;
469                }
470                Ok(())
471            }
472            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
473            Self::RocksDB(batch) => batch.delete::<tables::TransactionHashNumbers>(hash),
474        }
475    }
476}
477
478impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
479where
480    CURSOR: DbCursorRW<tables::StoragesHistory> + DbCursorRO<tables::StoragesHistory>,
481{
482    /// Puts a storage history entry.
483    pub fn put_storage_history(
484        &mut self,
485        key: StorageShardedKey,
486        value: &BlockNumberList,
487    ) -> ProviderResult<()> {
488        match self {
489            Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
490            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
491            Self::RocksDB(batch) => batch.put::<tables::StoragesHistory>(key, value),
492        }
493    }
494
495    /// Deletes a storage history entry.
496    pub fn delete_storage_history(&mut self, key: StorageShardedKey) -> ProviderResult<()> {
497        match self {
498            Self::Database(cursor) => {
499                if cursor.seek_exact(key)?.is_some() {
500                    cursor.delete_current()?;
501                }
502                Ok(())
503            }
504            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
505            Self::RocksDB(batch) => batch.delete::<tables::StoragesHistory>(key),
506        }
507    }
508
509    /// Appends a storage history entry (for first sync - more efficient).
510    pub fn append_storage_history(
511        &mut self,
512        key: StorageShardedKey,
513        value: &BlockNumberList,
514    ) -> ProviderResult<()> {
515        match self {
516            Self::Database(cursor) => Ok(cursor.append(key, value)?),
517            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
518            Self::RocksDB(batch) => batch.put::<tables::StoragesHistory>(key, value),
519        }
520    }
521
522    /// Upserts a storage history entry (for incremental sync).
523    pub fn upsert_storage_history(
524        &mut self,
525        key: StorageShardedKey,
526        value: &BlockNumberList,
527    ) -> ProviderResult<()> {
528        match self {
529            Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
530            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
531            Self::RocksDB(batch) => batch.put::<tables::StoragesHistory>(key, value),
532        }
533    }
534
535    /// Gets the last shard for an address and storage key (keyed with `u64::MAX`).
536    pub fn get_last_storage_history_shard(
537        &mut self,
538        address: Address,
539        storage_key: B256,
540    ) -> ProviderResult<Option<BlockNumberList>> {
541        let key = StorageShardedKey::last(address, storage_key);
542        match self {
543            Self::Database(cursor) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
544            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
545            Self::RocksDB(batch) => batch.get::<tables::StoragesHistory>(key),
546        }
547    }
548}
549
550impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
551where
552    CURSOR: DbCursorRW<tables::AccountsHistory> + DbCursorRO<tables::AccountsHistory>,
553{
554    /// Appends an account history entry (for first sync - more efficient).
555    pub fn append_account_history(
556        &mut self,
557        key: ShardedKey<Address>,
558        value: &BlockNumberList,
559    ) -> ProviderResult<()> {
560        match self {
561            Self::Database(cursor) => Ok(cursor.append(key, value)?),
562            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
563            Self::RocksDB(batch) => batch.put::<tables::AccountsHistory>(key, value),
564        }
565    }
566
567    /// Upserts an account history entry (for incremental sync).
568    pub fn upsert_account_history(
569        &mut self,
570        key: ShardedKey<Address>,
571        value: &BlockNumberList,
572    ) -> ProviderResult<()> {
573        match self {
574            Self::Database(cursor) => Ok(cursor.upsert(key, value)?),
575            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
576            Self::RocksDB(batch) => batch.put::<tables::AccountsHistory>(key, value),
577        }
578    }
579
580    /// Gets the last shard for an address (keyed with `u64::MAX`).
581    pub fn get_last_account_history_shard(
582        &mut self,
583        address: Address,
584    ) -> ProviderResult<Option<BlockNumberList>> {
585        match self {
586            Self::Database(cursor) => {
587                Ok(cursor.seek_exact(ShardedKey::last(address))?.map(|(_, v)| v))
588            }
589            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
590            Self::RocksDB(batch) => batch.get::<tables::AccountsHistory>(ShardedKey::last(address)),
591        }
592    }
593
594    /// Deletes an account history entry.
595    pub fn delete_account_history(&mut self, key: ShardedKey<Address>) -> ProviderResult<()> {
596        match self {
597            Self::Database(cursor) => {
598                if cursor.seek_exact(key)?.is_some() {
599                    cursor.delete_current()?;
600                }
601                Ok(())
602            }
603            Self::StaticFile(_) => Err(ProviderError::UnsupportedProvider),
604            Self::RocksDB(batch) => batch.delete::<tables::AccountsHistory>(key),
605        }
606    }
607}
608
609impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
610where
611    CURSOR: DbDupCursorRW<tables::AccountChangeSets>,
612{
613    /// Append account changeset for a block.
614    ///
615    /// NOTE: This _sorts_ the changesets by address before appending
616    pub fn append_account_changeset(
617        &mut self,
618        block_number: BlockNumber,
619        mut changeset: Vec<AccountBeforeTx>,
620    ) -> ProviderResult<()> {
621        // First sort the changesets
622        changeset.par_sort_by_key(|a| a.address);
623        match self {
624            Self::Database(cursor) => {
625                for change in changeset {
626                    cursor.append_dup(block_number, change)?;
627                }
628            }
629            Self::StaticFile(writer) => {
630                writer.append_account_changeset(changeset, block_number)?;
631            }
632            Self::RocksDB(_) => return Err(ProviderError::UnsupportedProvider),
633        }
634
635        Ok(())
636    }
637}
638
639impl<'a, CURSOR, N: NodePrimitives> EitherWriter<'a, CURSOR, N>
640where
641    CURSOR: DbDupCursorRW<tables::StorageChangeSets>,
642{
643    /// Append storage changeset for a block.
644    ///
645    /// NOTE: This _sorts_ the changesets by address and storage key before appending.
646    pub fn append_storage_changeset(
647        &mut self,
648        block_number: BlockNumber,
649        mut changeset: Vec<StorageBeforeTx>,
650    ) -> ProviderResult<()> {
651        changeset.par_sort_by_key(|change| (change.address, change.key));
652
653        match self {
654            Self::Database(cursor) => {
655                for change in changeset {
656                    let storage_id = BlockNumberAddress((block_number, change.address));
657                    cursor.append_dup(
658                        storage_id,
659                        StorageEntry { key: change.key, value: change.value },
660                    )?;
661                }
662            }
663            Self::StaticFile(writer) => {
664                writer.append_storage_changeset(changeset, block_number)?;
665            }
666            Self::RocksDB(_) => return Err(ProviderError::UnsupportedProvider),
667        }
668
669        Ok(())
670    }
671}
672
673/// Represents a source for reading data, either from database, static files, or `RocksDB`.
674#[derive(Debug, Display)]
675pub enum EitherReader<'a, CURSOR, N> {
676    /// Read from database table via cursor
677    Database(CURSOR, PhantomData<&'a ()>),
678    /// Read from static file
679    StaticFile(StaticFileProvider<N>, PhantomData<&'a ()>),
680    /// Read from `RocksDB` snapshot (works in both read-only and read-write modes)
681    RocksDB(crate::providers::rocksdb::RocksReadSnapshot<'a>),
682}
683
684impl<'a> EitherReader<'a, (), ()> {
685    /// Creates a new [`EitherReader`] for senders based on storage settings.
686    pub fn new_senders<P>(
687        provider: &P,
688    ) -> ProviderResult<EitherReaderTy<'a, P, tables::TransactionSenders>>
689    where
690        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache + StaticFileProviderFactory,
691        P::Tx: DbTx,
692    {
693        if EitherWriterDestination::senders(provider).is_static_file() {
694            Ok(EitherReader::StaticFile(provider.static_file_provider(), PhantomData))
695        } else {
696            Ok(EitherReader::Database(
697                provider.tx_ref().cursor_read::<tables::TransactionSenders>()?,
698                PhantomData,
699            ))
700        }
701    }
702
703    /// Creates a new [`EitherReader`] for storages history based on storage settings.
704    pub fn new_storages_history<P>(
705        provider: &P,
706        rocksdb: RocksDBRefArg<'a>,
707    ) -> ProviderResult<EitherReaderTy<'a, P, tables::StoragesHistory>>
708    where
709        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
710        P::Tx: DbTx,
711    {
712        if provider.cached_storage_settings().storage_v2 {
713            return Ok(EitherReader::RocksDB(
714                rocksdb.expect("storages_history_in_rocksdb requires rocksdb snapshot"),
715            ));
716        }
717
718        Ok(EitherReader::Database(
719            provider.tx_ref().cursor_read::<tables::StoragesHistory>()?,
720            PhantomData,
721        ))
722    }
723
724    /// Creates a new [`EitherReader`] for transaction hash numbers based on storage settings.
725    pub fn new_transaction_hash_numbers<P>(
726        provider: &P,
727        rocksdb: RocksDBRefArg<'a>,
728    ) -> ProviderResult<EitherReaderTy<'a, P, tables::TransactionHashNumbers>>
729    where
730        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
731        P::Tx: DbTx,
732    {
733        if provider.cached_storage_settings().storage_v2 {
734            return Ok(EitherReader::RocksDB(
735                rocksdb.expect("transaction_hash_numbers_in_rocksdb requires rocksdb snapshot"),
736            ));
737        }
738
739        Ok(EitherReader::Database(
740            provider.tx_ref().cursor_read::<tables::TransactionHashNumbers>()?,
741            PhantomData,
742        ))
743    }
744
745    /// Creates a new [`EitherReader`] for account history based on storage settings.
746    pub fn new_accounts_history<P>(
747        provider: &P,
748        rocksdb: RocksDBRefArg<'a>,
749    ) -> ProviderResult<EitherReaderTy<'a, P, tables::AccountsHistory>>
750    where
751        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache,
752        P::Tx: DbTx,
753    {
754        if provider.cached_storage_settings().storage_v2 {
755            return Ok(EitherReader::RocksDB(
756                rocksdb.expect("account_history_in_rocksdb requires rocksdb snapshot"),
757            ));
758        }
759
760        Ok(EitherReader::Database(
761            provider.tx_ref().cursor_read::<tables::AccountsHistory>()?,
762            PhantomData,
763        ))
764    }
765
766    /// Creates a new [`EitherReader`] for account changesets based on storage settings.
767    pub fn new_account_changesets<P>(
768        provider: &P,
769    ) -> ProviderResult<DupEitherReaderTy<'a, P, tables::AccountChangeSets>>
770    where
771        P: DBProvider + NodePrimitivesProvider + StorageSettingsCache + StaticFileProviderFactory,
772        P::Tx: DbTx,
773    {
774        if EitherWriterDestination::account_changesets(provider).is_static_file() {
775            Ok(EitherReader::StaticFile(provider.static_file_provider(), PhantomData))
776        } else {
777            Ok(EitherReader::Database(
778                provider.tx_ref().cursor_dup_read::<tables::AccountChangeSets>()?,
779                PhantomData,
780            ))
781        }
782    }
783}
784
785impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
786where
787    CURSOR: DbCursorRO<tables::TransactionSenders>,
788{
789    /// Fetches the senders for a range of transactions.
790    pub fn senders_by_tx_range(
791        &mut self,
792        range: Range<TxNumber>,
793    ) -> ProviderResult<HashMap<TxNumber, Address>> {
794        match self {
795            Self::Database(cursor, _) => cursor
796                .walk_range(range)?
797                .map(|result| result.map_err(ProviderError::from))
798                .collect::<ProviderResult<HashMap<_, _>>>(),
799            Self::StaticFile(provider, _) => range
800                .clone()
801                .zip(provider.fetch_range_iter(
802                    StaticFileSegment::TransactionSenders,
803                    range,
804                    |cursor, number| cursor.get_one::<TransactionSenderMask>(number.into()),
805                )?)
806                .filter_map(|(tx_num, sender)| {
807                    let result = sender.transpose()?;
808                    Some(result.map(|sender| (tx_num, sender)))
809                })
810                .collect::<ProviderResult<HashMap<_, _>>>(),
811            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
812        }
813    }
814}
815
816impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
817where
818    CURSOR: DbCursorRO<tables::TransactionHashNumbers>,
819{
820    /// Gets a transaction number by its hash.
821    pub fn get_transaction_hash_number(
822        &mut self,
823        hash: TxHash,
824    ) -> ProviderResult<Option<TxNumber>> {
825        match self {
826            Self::Database(cursor, _) => Ok(cursor.seek_exact(hash)?.map(|(_, v)| v)),
827            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
828            Self::RocksDB(snapshot) => snapshot.get::<tables::TransactionHashNumbers>(hash),
829        }
830    }
831}
832
833impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
834where
835    CURSOR: DbCursorRO<tables::StoragesHistory>,
836{
837    /// Gets a storage history shard entry for the given [`StorageShardedKey`], if present.
838    pub fn get_storage_history(
839        &mut self,
840        key: StorageShardedKey,
841    ) -> ProviderResult<Option<BlockNumberList>> {
842        match self {
843            Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
844            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
845            Self::RocksDB(snapshot) => snapshot.get::<tables::StoragesHistory>(key),
846        }
847    }
848
849    /// Lookup storage history and return [`HistoryInfo`].
850    pub fn storage_history_info(
851        &mut self,
852        address: Address,
853        storage_key: alloy_primitives::B256,
854        block_number: BlockNumber,
855        lowest_available_block_number: Option<BlockNumber>,
856        visible_tip: BlockNumber,
857    ) -> ProviderResult<HistoryInfo> {
858        match self {
859            Self::Database(cursor, _) => {
860                let key = StorageShardedKey::new(address, storage_key, block_number);
861                history_info::<tables::StoragesHistory, _, _>(
862                    cursor,
863                    key,
864                    block_number,
865                    |k| k.address == address && k.sharded_key.key == storage_key,
866                    lowest_available_block_number,
867                )
868            }
869            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
870            Self::RocksDB(snapshot) => snapshot.storage_history_info(
871                address,
872                storage_key,
873                block_number,
874                lowest_available_block_number,
875                visible_tip,
876            ),
877        }
878    }
879}
880
881impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
882where
883    CURSOR: DbCursorRO<tables::AccountsHistory>,
884{
885    /// Gets an account history shard entry for the given [`ShardedKey`], if present.
886    pub fn get_account_history(
887        &mut self,
888        key: ShardedKey<Address>,
889    ) -> ProviderResult<Option<BlockNumberList>> {
890        match self {
891            Self::Database(cursor, _) => Ok(cursor.seek_exact(key)?.map(|(_, v)| v)),
892            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
893            Self::RocksDB(snapshot) => snapshot.get::<tables::AccountsHistory>(key),
894        }
895    }
896
897    /// Lookup account history and return [`HistoryInfo`].
898    pub fn account_history_info(
899        &mut self,
900        address: Address,
901        block_number: BlockNumber,
902        lowest_available_block_number: Option<BlockNumber>,
903        visible_tip: BlockNumber,
904    ) -> ProviderResult<HistoryInfo> {
905        match self {
906            Self::Database(cursor, _) => {
907                let key = ShardedKey::new(address, block_number);
908                history_info::<tables::AccountsHistory, _, _>(
909                    cursor,
910                    key,
911                    block_number,
912                    |k| k.key == address,
913                    lowest_available_block_number,
914                )
915            }
916            Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
917            Self::RocksDB(snapshot) => snapshot.account_history_info(
918                address,
919                block_number,
920                lowest_available_block_number,
921                visible_tip,
922            ),
923        }
924    }
925}
926
927impl<CURSOR, N: NodePrimitives> EitherReader<'_, CURSOR, N>
928where
929    CURSOR: DbCursorRO<tables::AccountChangeSets>,
930{
931    /// Iterate over account changesets and return all account address that were changed.
932    pub fn changed_accounts_with_range(
933        &mut self,
934        range: RangeInclusive<BlockNumber>,
935    ) -> ProviderResult<BTreeSet<Address>> {
936        match self {
937            Self::StaticFile(provider, _) => {
938                let highest_static_block =
939                    provider.get_highest_static_file_block(StaticFileSegment::AccountChangeSets);
940
941                let Some(highest) = highest_static_block else {
942                    return Err(ProviderError::MissingHighestStaticFileBlock(
943                        StaticFileSegment::AccountChangeSets,
944                    ))
945                };
946
947                let start = *range.start();
948                let static_end = (*range.end()).min(highest);
949
950                let mut changed_accounts = BTreeSet::default();
951                if start <= static_end {
952                    for block in start..=static_end {
953                        let block_changesets = provider.account_block_changeset(block)?;
954                        for changeset in block_changesets {
955                            changed_accounts.insert(changeset.address);
956                        }
957                    }
958                }
959
960                Ok(changed_accounts)
961            }
962            Self::Database(provider, _) => provider
963                .walk_range(range)?
964                .map(|entry| {
965                    entry.map(|(_, account_before)| account_before.address).map_err(Into::into)
966                })
967                .collect(),
968            Self::RocksDB(_) => Err(ProviderError::UnsupportedProvider),
969        }
970    }
971}
972
973/// Destination for writing data.
974#[derive(Debug, EnumIs)]
975pub enum EitherWriterDestination {
976    /// Write to database table
977    Database,
978    /// Write to static file
979    StaticFile,
980    /// Write to `RocksDB`
981    RocksDB,
982}
983
984impl EitherWriterDestination {
985    /// Returns the destination for writing senders based on storage settings.
986    pub fn senders<P>(provider: &P) -> Self
987    where
988        P: StorageSettingsCache,
989    {
990        // Write senders to static files only if they're explicitly enabled
991        if provider.cached_storage_settings().storage_v2 {
992            Self::StaticFile
993        } else {
994            Self::Database
995        }
996    }
997
998    /// Returns the destination for writing account changesets based on storage settings.
999    pub fn account_changesets<P>(provider: &P) -> Self
1000    where
1001        P: StorageSettingsCache,
1002    {
1003        // Write account changesets to static files only if they're explicitly enabled
1004        if provider.cached_storage_settings().storage_v2 {
1005            Self::StaticFile
1006        } else {
1007            Self::Database
1008        }
1009    }
1010
1011    /// Returns the destination for writing storage changesets based on storage settings.
1012    pub fn storage_changesets<P>(provider: &P) -> Self
1013    where
1014        P: StorageSettingsCache,
1015    {
1016        // Write storage changesets to static files only if they're explicitly enabled
1017        if provider.cached_storage_settings().storage_v2 {
1018            Self::StaticFile
1019        } else {
1020            Self::Database
1021        }
1022    }
1023}
1024
1025#[cfg(test)]
1026mod tests {
1027    use crate::{test_utils::create_test_provider_factory, StaticFileWriter};
1028
1029    use super::*;
1030    use alloy_primitives::Address;
1031    use reth_db::models::AccountBeforeTx;
1032    use reth_static_file_types::StaticFileSegment;
1033    use reth_storage_api::{DatabaseProviderFactory, StorageSettings};
1034
1035    /// Verifies that `changed_accounts_with_range` correctly caps the query range to the
1036    /// static file tip when the requested range extends beyond it.
1037    ///
1038    /// This test documents the fix for an off-by-one bug where the code computed
1039    /// `static_end = range.end().min(highest + 1)` instead of `min(highest)`.
1040    /// The bug allowed iteration to attempt reading block `highest + 1` which doesn't
1041    /// exist (silently returning empty results due to `MissingStaticFileBlock` handling).
1042    ///
1043    /// While the bug was masked by error handling, it caused:
1044    /// 1. Unnecessary iteration/lookup for non-existent blocks
1045    /// 2. Potential overflow when `highest == u64::MAX`
1046    #[test]
1047    fn test_changed_accounts_with_range_caps_at_static_file_tip() {
1048        let factory = create_test_provider_factory();
1049        let highest_block = 5u64;
1050
1051        let addresses: Vec<Address> = (0..=highest_block)
1052            .map(|i| {
1053                let mut addr = Address::ZERO;
1054                addr.0[0] = i as u8;
1055                addr
1056            })
1057            .collect();
1058
1059        {
1060            let sf_provider = factory.static_file_provider();
1061            let mut writer =
1062                sf_provider.latest_writer(StaticFileSegment::AccountChangeSets).unwrap();
1063
1064            for block_num in 0..=highest_block {
1065                let changeset =
1066                    vec![AccountBeforeTx { address: addresses[block_num as usize], info: None }];
1067                writer.append_account_changeset(changeset, block_num).unwrap();
1068            }
1069            writer.commit().unwrap();
1070        }
1071
1072        factory.set_storage_settings_cache(StorageSettings::v2());
1073
1074        let provider = factory.database_provider_ro().unwrap();
1075
1076        let sf_tip = provider
1077            .static_file_provider()
1078            .get_highest_static_file_block(StaticFileSegment::AccountChangeSets);
1079        assert_eq!(sf_tip, Some(highest_block));
1080
1081        let mut reader = EitherReader::new_account_changesets(&provider).unwrap();
1082        assert!(matches!(reader, EitherReader::StaticFile(_, _)));
1083
1084        // Query range 0..=10 when tip is 5 - should return only accounts from blocks 0-5
1085        let result = reader.changed_accounts_with_range(0..=10).unwrap();
1086
1087        let expected: BTreeSet<Address> = addresses.into_iter().collect();
1088        assert_eq!(result, expected);
1089    }
1090
1091    #[test]
1092    fn test_reader_senders_by_tx_range() {
1093        let factory = create_test_provider_factory();
1094
1095        // Insert senders only from 1 to 4, but we will query from 0 to 5.
1096        let senders = [
1097            (1, Address::random()),
1098            (2, Address::random()),
1099            (3, Address::random()),
1100            (4, Address::random()),
1101        ];
1102
1103        for transaction_senders_in_static_files in [false, true] {
1104            factory.set_storage_settings_cache(if transaction_senders_in_static_files {
1105                StorageSettings::v2()
1106            } else {
1107                StorageSettings::v1()
1108            });
1109
1110            let provider = factory.database_provider_rw().unwrap();
1111            let mut writer = EitherWriter::new_senders(&provider, 0).unwrap();
1112            if transaction_senders_in_static_files {
1113                assert!(matches!(writer, EitherWriter::StaticFile(_)));
1114            } else {
1115                assert!(matches!(writer, EitherWriter::Database(_)));
1116            }
1117
1118            writer.increment_block(0).unwrap();
1119            writer.append_senders(senders.iter().copied()).unwrap();
1120            drop(writer);
1121            provider.commit().unwrap();
1122
1123            let provider = factory.database_provider_ro().unwrap();
1124            let mut reader = EitherReader::new_senders(&provider).unwrap();
1125            if transaction_senders_in_static_files {
1126                assert!(matches!(reader, EitherReader::StaticFile(_, _)));
1127            } else {
1128                assert!(matches!(reader, EitherReader::Database(_, _)));
1129            }
1130
1131            assert_eq!(
1132                reader.senders_by_tx_range(0..6).unwrap(),
1133                senders.iter().copied().collect::<HashMap<_, _>>(),
1134                "{reader}"
1135            );
1136        }
1137    }
1138}
1139
1140#[cfg(test)]
1141mod rocksdb_tests {
1142    use super::*;
1143    use crate::{
1144        providers::rocksdb::{RocksDBBuilder, RocksDBProvider},
1145        test_utils::create_test_provider_factory,
1146        RocksDBProviderFactory,
1147    };
1148    use alloy_primitives::{Address, B256};
1149    use reth_db_api::{
1150        models::{storage_sharded_key::StorageShardedKey, IntegerList, ShardedKey},
1151        tables,
1152        transaction::DbTxMut,
1153    };
1154    use reth_ethereum_primitives::EthPrimitives;
1155    use reth_storage_api::{DatabaseProviderFactory, StorageSettings};
1156    use std::marker::PhantomData;
1157    use tempfile::TempDir;
1158
1159    fn create_rocksdb_provider() -> (TempDir, RocksDBProvider) {
1160        let temp_dir = TempDir::new().unwrap();
1161        let provider = RocksDBBuilder::new(temp_dir.path())
1162            .with_table::<tables::TransactionHashNumbers>()
1163            .with_table::<tables::StoragesHistory>()
1164            .with_table::<tables::AccountsHistory>()
1165            .build()
1166            .unwrap();
1167        (temp_dir, provider)
1168    }
1169
1170    /// Test that `EitherWriter::new_transaction_hash_numbers` creates a `RocksDB` writer
1171    /// when the storage setting is enabled, and that put operations followed by commit
1172    /// persist the data to `RocksDB`.
1173    #[test]
1174    fn test_either_writer_transaction_hash_numbers_with_rocksdb() {
1175        let factory = create_test_provider_factory();
1176
1177        // Enable RocksDB for transaction hash numbers
1178        factory.set_storage_settings_cache(StorageSettings::v2());
1179
1180        let hash1 = B256::from([1u8; 32]);
1181        let hash2 = B256::from([2u8; 32]);
1182        let tx_num1 = 100u64;
1183        let tx_num2 = 200u64;
1184
1185        // Get the RocksDB batch from the provider
1186        let rocksdb = factory.rocksdb_provider();
1187        let batch = rocksdb.batch();
1188
1189        // Create EitherWriter with RocksDB
1190        let provider = factory.database_provider_rw().unwrap();
1191        let mut writer = EitherWriter::new_transaction_hash_numbers(&provider, batch).unwrap();
1192
1193        // Verify we got a RocksDB writer
1194        assert!(matches!(writer, EitherWriter::RocksDB(_)));
1195
1196        // Write transaction hash numbers (append_only=false since we're using RocksDB)
1197        writer.put_transaction_hash_number(hash1, tx_num1, false).unwrap();
1198        writer.put_transaction_hash_number(hash2, tx_num2, false).unwrap();
1199
1200        // Extract the batch and register with provider for commit
1201        if let Some(batch) = writer.into_raw_rocksdb_batch() {
1202            provider.set_pending_rocksdb_batch(batch);
1203        }
1204
1205        // Commit via provider - this commits RocksDB batch too
1206        provider.commit().unwrap();
1207
1208        // Verify data was written to RocksDB
1209        let rocksdb = factory.rocksdb_provider();
1210        assert_eq!(rocksdb.get::<tables::TransactionHashNumbers>(hash1).unwrap(), Some(tx_num1));
1211        assert_eq!(rocksdb.get::<tables::TransactionHashNumbers>(hash2).unwrap(), Some(tx_num2));
1212    }
1213
1214    /// Test that `EitherWriter::delete_transaction_hash_number` works with `RocksDB`.
1215    #[test]
1216    fn test_either_writer_delete_transaction_hash_number_with_rocksdb() {
1217        let factory = create_test_provider_factory();
1218
1219        // Enable RocksDB for transaction hash numbers
1220        factory.set_storage_settings_cache(StorageSettings::v2());
1221
1222        let hash = B256::from([1u8; 32]);
1223        let tx_num = 100u64;
1224
1225        // First, write a value directly to RocksDB
1226        let rocksdb = factory.rocksdb_provider();
1227        rocksdb.put::<tables::TransactionHashNumbers>(hash, &tx_num).unwrap();
1228        assert_eq!(rocksdb.get::<tables::TransactionHashNumbers>(hash).unwrap(), Some(tx_num));
1229
1230        // Now delete using EitherWriter
1231        let batch = rocksdb.batch();
1232        let provider = factory.database_provider_rw().unwrap();
1233        let mut writer = EitherWriter::new_transaction_hash_numbers(&provider, batch).unwrap();
1234        writer.delete_transaction_hash_number(hash).unwrap();
1235
1236        // Extract the batch and commit via provider
1237        if let Some(batch) = writer.into_raw_rocksdb_batch() {
1238            provider.set_pending_rocksdb_batch(batch);
1239        }
1240        provider.commit().unwrap();
1241
1242        // Verify deletion
1243        let rocksdb = factory.rocksdb_provider();
1244        assert_eq!(rocksdb.get::<tables::TransactionHashNumbers>(hash).unwrap(), None);
1245    }
1246
1247    #[test]
1248    fn test_rocksdb_batch_transaction_hash_numbers() {
1249        let (_temp_dir, provider) = create_rocksdb_provider();
1250
1251        let hash1 = B256::from([1u8; 32]);
1252        let hash2 = B256::from([2u8; 32]);
1253        let tx_num1 = 100u64;
1254        let tx_num2 = 200u64;
1255
1256        // Write via RocksDBBatch (same as EitherWriter::RocksDB would use internally)
1257        let mut batch = provider.batch();
1258        batch.put::<tables::TransactionHashNumbers>(hash1, &tx_num1).unwrap();
1259        batch.put::<tables::TransactionHashNumbers>(hash2, &tx_num2).unwrap();
1260        batch.commit().unwrap();
1261
1262        // Read via RocksTx (same as EitherReader::RocksDB would use internally)
1263        let tx = provider.tx();
1264        assert_eq!(tx.get::<tables::TransactionHashNumbers>(hash1).unwrap(), Some(tx_num1));
1265        assert_eq!(tx.get::<tables::TransactionHashNumbers>(hash2).unwrap(), Some(tx_num2));
1266
1267        // Test missing key
1268        let missing_hash = B256::from([99u8; 32]);
1269        assert_eq!(tx.get::<tables::TransactionHashNumbers>(missing_hash).unwrap(), None);
1270    }
1271
1272    #[test]
1273    fn test_rocksdb_batch_storage_history() {
1274        let (_temp_dir, provider) = create_rocksdb_provider();
1275
1276        let address = Address::random();
1277        let storage_key = B256::from([1u8; 32]);
1278        let key = StorageShardedKey::new(address, storage_key, 1000);
1279        let value = IntegerList::new([1, 5, 10, 50]).unwrap();
1280
1281        // Write via RocksDBBatch
1282        let mut batch = provider.batch();
1283        batch.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
1284        batch.commit().unwrap();
1285
1286        // Read via RocksTx
1287        let tx = provider.tx();
1288        let result = tx.get::<tables::StoragesHistory>(key).unwrap();
1289        assert_eq!(result, Some(value));
1290
1291        // Test missing key
1292        let missing_key = StorageShardedKey::new(Address::random(), B256::ZERO, 0);
1293        assert_eq!(tx.get::<tables::StoragesHistory>(missing_key).unwrap(), None);
1294    }
1295
1296    #[test]
1297    fn test_rocksdb_batch_account_history() {
1298        let (_temp_dir, provider) = create_rocksdb_provider();
1299
1300        let address = Address::random();
1301        let key = ShardedKey::new(address, 1000);
1302        let value = IntegerList::new([1, 10, 100, 500]).unwrap();
1303
1304        // Write via RocksDBBatch
1305        let mut batch = provider.batch();
1306        batch.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
1307        batch.commit().unwrap();
1308
1309        // Read via RocksTx
1310        let tx = provider.tx();
1311        let result = tx.get::<tables::AccountsHistory>(key).unwrap();
1312        assert_eq!(result, Some(value));
1313
1314        // Test missing key
1315        let missing_key = ShardedKey::new(Address::random(), 0);
1316        assert_eq!(tx.get::<tables::AccountsHistory>(missing_key).unwrap(), None);
1317    }
1318
1319    #[test]
1320    fn test_rocksdb_batch_delete_transaction_hash_number() {
1321        let (_temp_dir, provider) = create_rocksdb_provider();
1322
1323        let hash = B256::from([1u8; 32]);
1324        let tx_num = 100u64;
1325
1326        // First write
1327        provider.put::<tables::TransactionHashNumbers>(hash, &tx_num).unwrap();
1328        assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), Some(tx_num));
1329
1330        // Delete via RocksDBBatch
1331        let mut batch = provider.batch();
1332        batch.delete::<tables::TransactionHashNumbers>(hash).unwrap();
1333        batch.commit().unwrap();
1334
1335        // Verify deletion
1336        assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), None);
1337    }
1338
1339    #[test]
1340    fn test_rocksdb_batch_delete_storage_history() {
1341        let (_temp_dir, provider) = create_rocksdb_provider();
1342
1343        let address = Address::random();
1344        let storage_key = B256::from([1u8; 32]);
1345        let key = StorageShardedKey::new(address, storage_key, 1000);
1346        let value = IntegerList::new([1, 5, 10]).unwrap();
1347
1348        // First write
1349        provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
1350        assert!(provider.get::<tables::StoragesHistory>(key.clone()).unwrap().is_some());
1351
1352        // Delete via RocksDBBatch
1353        let mut batch = provider.batch();
1354        batch.delete::<tables::StoragesHistory>(key.clone()).unwrap();
1355        batch.commit().unwrap();
1356
1357        // Verify deletion
1358        assert_eq!(provider.get::<tables::StoragesHistory>(key).unwrap(), None);
1359    }
1360
1361    #[test]
1362    fn test_rocksdb_batch_delete_account_history() {
1363        let (_temp_dir, provider) = create_rocksdb_provider();
1364
1365        let address = Address::random();
1366        let key = ShardedKey::new(address, 1000);
1367        let value = IntegerList::new([1, 10, 100]).unwrap();
1368
1369        // First write
1370        provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
1371        assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
1372
1373        // Delete via RocksDBBatch
1374        let mut batch = provider.batch();
1375        batch.delete::<tables::AccountsHistory>(key.clone()).unwrap();
1376        batch.commit().unwrap();
1377
1378        // Verify deletion
1379        assert_eq!(provider.get::<tables::AccountsHistory>(key).unwrap(), None);
1380    }
1381
1382    // ==================== Parametrized Backend Equivalence Tests ====================
1383    //
1384    // These tests verify that MDBX and RocksDB produce identical results for history lookups.
1385    // Each scenario sets up the same data in both backends and asserts identical HistoryInfo.
1386
1387    /// Query parameters for a history lookup test case.
1388    struct HistoryQuery {
1389        block_number: BlockNumber,
1390        lowest_available: Option<BlockNumber>,
1391        expected: HistoryInfo,
1392    }
1393
1394    // Type aliases for cursor types (needed for EitherWriter/EitherReader type inference)
1395    type AccountsHistoryWriteCursor =
1396        reth_db::mdbx::cursor::Cursor<reth_db::mdbx::RW, tables::AccountsHistory>;
1397    type StoragesHistoryWriteCursor =
1398        reth_db::mdbx::cursor::Cursor<reth_db::mdbx::RW, tables::StoragesHistory>;
1399    type AccountsHistoryReadCursor =
1400        reth_db::mdbx::cursor::Cursor<reth_db::mdbx::RO, tables::AccountsHistory>;
1401    type StoragesHistoryReadCursor =
1402        reth_db::mdbx::cursor::Cursor<reth_db::mdbx::RO, tables::StoragesHistory>;
1403
1404    /// Runs the same account history queries against both MDBX and `RocksDB` backends,
1405    /// asserting they produce identical results.
1406    fn run_account_history_scenario(
1407        scenario_name: &str,
1408        address: Address,
1409        shards: &[(BlockNumber, Vec<BlockNumber>)], // (shard_highest_block, blocks_in_shard)
1410        queries: &[HistoryQuery],
1411    ) {
1412        // Setup MDBX and RocksDB with identical data using EitherWriter
1413        let factory = create_test_provider_factory();
1414        let mdbx_provider = factory.database_provider_rw().unwrap();
1415        let (temp_dir, rocks_provider) = create_rocksdb_provider();
1416
1417        // Create writers for both backends
1418        let mut mdbx_writer: EitherWriter<'_, AccountsHistoryWriteCursor, EthPrimitives> =
1419            EitherWriter::Database(
1420                mdbx_provider.tx_ref().cursor_write::<tables::AccountsHistory>().unwrap(),
1421            );
1422        let mut rocks_writer: EitherWriter<'_, AccountsHistoryWriteCursor, EthPrimitives> =
1423            EitherWriter::RocksDB(rocks_provider.batch());
1424
1425        // Write identical data to both backends in a single loop
1426        for (highest_block, blocks) in shards {
1427            let key = ShardedKey::new(address, *highest_block);
1428            let value = IntegerList::new(blocks.clone()).unwrap();
1429            mdbx_writer.upsert_account_history(key.clone(), &value).unwrap();
1430            rocks_writer.upsert_account_history(key, &value).unwrap();
1431        }
1432
1433        // Commit both backends
1434        drop(mdbx_writer);
1435        mdbx_provider.commit().unwrap();
1436        if let EitherWriter::RocksDB(batch) = rocks_writer {
1437            batch.commit().unwrap();
1438        }
1439
1440        // Run queries against both backends using EitherReader
1441        let mdbx_ro = factory.database_provider_ro().unwrap();
1442        let rocks_snapshot = rocks_provider.snapshot();
1443
1444        for (i, query) in queries.iter().enumerate() {
1445            // MDBX query via EitherReader
1446            let mut mdbx_reader: EitherReader<'_, AccountsHistoryReadCursor, EthPrimitives> =
1447                EitherReader::Database(
1448                    mdbx_ro.tx_ref().cursor_read::<tables::AccountsHistory>().unwrap(),
1449                    PhantomData,
1450                );
1451            let mdbx_result = mdbx_reader
1452                .account_history_info(address, query.block_number, query.lowest_available, u64::MAX)
1453                .unwrap();
1454
1455            // RocksDB query via EitherReader — reuse snapshot for consistent view
1456            let rocks_result = rocks_snapshot
1457                .account_history_info(address, query.block_number, query.lowest_available, u64::MAX)
1458                .unwrap();
1459
1460            // Assert both backends produce identical results
1461            assert_eq!(
1462                mdbx_result,
1463                rocks_result,
1464                "Backend mismatch in scenario '{}' query {}: block={}, lowest={:?}\n\
1465                 MDBX: {:?}, RocksDB: {:?}",
1466                scenario_name,
1467                i,
1468                query.block_number,
1469                query.lowest_available,
1470                mdbx_result,
1471                rocks_result
1472            );
1473
1474            // Also verify against expected result
1475            assert_eq!(
1476                mdbx_result,
1477                query.expected,
1478                "Unexpected result in scenario '{}' query {}: block={}, lowest={:?}\n\
1479                 Got: {:?}, Expected: {:?}",
1480                scenario_name,
1481                i,
1482                query.block_number,
1483                query.lowest_available,
1484                mdbx_result,
1485                query.expected
1486            );
1487        }
1488
1489        drop(temp_dir);
1490    }
1491
1492    /// Runs the same storage history queries against both MDBX and `RocksDB` backends,
1493    /// asserting they produce identical results.
1494    fn run_storage_history_scenario(
1495        scenario_name: &str,
1496        address: Address,
1497        storage_key: B256,
1498        shards: &[(BlockNumber, Vec<BlockNumber>)], // (shard_highest_block, blocks_in_shard)
1499        queries: &[HistoryQuery],
1500    ) {
1501        // Setup MDBX and RocksDB with identical data using EitherWriter
1502        let factory = create_test_provider_factory();
1503        let mdbx_provider = factory.database_provider_rw().unwrap();
1504        let (temp_dir, rocks_provider) = create_rocksdb_provider();
1505
1506        // Create writers for both backends
1507        let mut mdbx_writer: EitherWriter<'_, StoragesHistoryWriteCursor, EthPrimitives> =
1508            EitherWriter::Database(
1509                mdbx_provider.tx_ref().cursor_write::<tables::StoragesHistory>().unwrap(),
1510            );
1511        let mut rocks_writer: EitherWriter<'_, StoragesHistoryWriteCursor, EthPrimitives> =
1512            EitherWriter::RocksDB(rocks_provider.batch());
1513
1514        // Write identical data to both backends in a single loop
1515        for (highest_block, blocks) in shards {
1516            let key = StorageShardedKey::new(address, storage_key, *highest_block);
1517            let value = IntegerList::new(blocks.clone()).unwrap();
1518            mdbx_writer.put_storage_history(key.clone(), &value).unwrap();
1519            rocks_writer.put_storage_history(key, &value).unwrap();
1520        }
1521
1522        // Commit both backends
1523        drop(mdbx_writer);
1524        mdbx_provider.commit().unwrap();
1525        if let EitherWriter::RocksDB(batch) = rocks_writer {
1526            batch.commit().unwrap();
1527        }
1528
1529        // Run queries against both backends using EitherReader
1530        let mdbx_ro = factory.database_provider_ro().unwrap();
1531        let rocks_snapshot = rocks_provider.snapshot();
1532
1533        for (i, query) in queries.iter().enumerate() {
1534            // MDBX query via EitherReader
1535            let mut mdbx_reader: EitherReader<'_, StoragesHistoryReadCursor, EthPrimitives> =
1536                EitherReader::Database(
1537                    mdbx_ro.tx_ref().cursor_read::<tables::StoragesHistory>().unwrap(),
1538                    PhantomData,
1539                );
1540            let mdbx_result = mdbx_reader
1541                .storage_history_info(
1542                    address,
1543                    storage_key,
1544                    query.block_number,
1545                    query.lowest_available,
1546                    u64::MAX,
1547                )
1548                .unwrap();
1549
1550            // RocksDB query via snapshot — reuse for consistent view
1551            let rocks_result = rocks_snapshot
1552                .storage_history_info(
1553                    address,
1554                    storage_key,
1555                    query.block_number,
1556                    query.lowest_available,
1557                    u64::MAX,
1558                )
1559                .unwrap();
1560
1561            // Assert both backends produce identical results
1562            assert_eq!(
1563                mdbx_result,
1564                rocks_result,
1565                "Backend mismatch in scenario '{}' query {}: block={}, lowest={:?}\n\
1566                 MDBX: {:?}, RocksDB: {:?}",
1567                scenario_name,
1568                i,
1569                query.block_number,
1570                query.lowest_available,
1571                mdbx_result,
1572                rocks_result
1573            );
1574
1575            // Also verify against expected result
1576            assert_eq!(
1577                mdbx_result,
1578                query.expected,
1579                "Unexpected result in scenario '{}' query {}: block={}, lowest={:?}\n\
1580                 Got: {:?}, Expected: {:?}",
1581                scenario_name,
1582                i,
1583                query.block_number,
1584                query.lowest_available,
1585                mdbx_result,
1586                query.expected
1587            );
1588        }
1589
1590        drop(temp_dir);
1591    }
1592
1593    /// Tests account history lookups across both MDBX and `RocksDB` backends.
1594    ///
1595    /// Covers the following scenarios from PR2's `RocksDB`-only tests:
1596    /// 1. Single shard - basic lookups within one shard
1597    /// 2. Multiple shards - `prev()` shard detection and transitions
1598    /// 3. No history - query address with no entries
1599    /// 4. Pruning boundary - `lowest_available` boundary behavior (block at/after boundary)
1600    #[test]
1601    fn test_account_history_info_both_backends() {
1602        let address = Address::from([0x42; 20]);
1603
1604        // Scenario 1: Single shard with blocks [100, 200, 300]
1605        run_account_history_scenario(
1606            "single_shard",
1607            address,
1608            &[(u64::MAX, vec![100, 200, 300])],
1609            &[
1610                // Before first entry -> NotYetWritten
1611                HistoryQuery {
1612                    block_number: 50,
1613                    lowest_available: None,
1614                    expected: HistoryInfo::NotYetWritten,
1615                },
1616                // Between entries -> InChangeset(next_write)
1617                HistoryQuery {
1618                    block_number: 150,
1619                    lowest_available: None,
1620                    expected: HistoryInfo::InChangeset(200),
1621                },
1622                // Exact match on entry -> InChangeset(same_block)
1623                HistoryQuery {
1624                    block_number: 300,
1625                    lowest_available: None,
1626                    expected: HistoryInfo::InChangeset(300),
1627                },
1628                // After last entry in last shard -> InPlainState
1629                HistoryQuery {
1630                    block_number: 500,
1631                    lowest_available: None,
1632                    expected: HistoryInfo::InPlainState,
1633                },
1634            ],
1635        );
1636
1637        // Scenario 2: Multiple shards - tests prev() shard detection
1638        run_account_history_scenario(
1639            "multiple_shards",
1640            address,
1641            &[
1642                (500, vec![100, 200, 300, 400, 500]), // First shard ends at 500
1643                (u64::MAX, vec![600, 700, 800]),      // Last shard
1644            ],
1645            &[
1646                // Before first shard, no prev -> NotYetWritten
1647                HistoryQuery {
1648                    block_number: 50,
1649                    lowest_available: None,
1650                    expected: HistoryInfo::NotYetWritten,
1651                },
1652                // Within first shard
1653                HistoryQuery {
1654                    block_number: 150,
1655                    lowest_available: None,
1656                    expected: HistoryInfo::InChangeset(200),
1657                },
1658                // Between shards - prev() should find first shard
1659                HistoryQuery {
1660                    block_number: 550,
1661                    lowest_available: None,
1662                    expected: HistoryInfo::InChangeset(600),
1663                },
1664                // After all entries
1665                HistoryQuery {
1666                    block_number: 900,
1667                    lowest_available: None,
1668                    expected: HistoryInfo::InPlainState,
1669                },
1670            ],
1671        );
1672
1673        // Scenario 3: No history for address
1674        let address_without_history = Address::from([0x43; 20]);
1675        run_account_history_scenario(
1676            "no_history",
1677            address_without_history,
1678            &[], // No shards for this address
1679            &[HistoryQuery {
1680                block_number: 150,
1681                lowest_available: None,
1682                expected: HistoryInfo::NotYetWritten,
1683            }],
1684        );
1685
1686        // Scenario 4: Query at pruning boundary
1687        // Note: We test block >= lowest_available because HistoricalStateProviderRef
1688        // errors on blocks below the pruning boundary before doing the lookup.
1689        // The RocksDB implementation doesn't have this check at the same level.
1690        // This tests that when pruning IS available, both backends agree.
1691        run_account_history_scenario(
1692            "with_pruning_boundary",
1693            address,
1694            &[(u64::MAX, vec![100, 200, 300])],
1695            &[
1696                // At pruning boundary -> InChangeset(first entry after block)
1697                HistoryQuery {
1698                    block_number: 100,
1699                    lowest_available: Some(100),
1700                    expected: HistoryInfo::InChangeset(100),
1701                },
1702                // After pruning boundary, between entries
1703                HistoryQuery {
1704                    block_number: 150,
1705                    lowest_available: Some(100),
1706                    expected: HistoryInfo::InChangeset(200),
1707                },
1708            ],
1709        );
1710    }
1711
1712    /// Tests storage history lookups across both MDBX and `RocksDB` backends.
1713    #[test]
1714    fn test_storage_history_info_both_backends() {
1715        let address = Address::from([0x42; 20]);
1716        let storage_key = B256::from([0x01; 32]);
1717        let other_storage_key = B256::from([0x02; 32]);
1718
1719        // Single shard with blocks [100, 200, 300]
1720        run_storage_history_scenario(
1721            "storage_single_shard",
1722            address,
1723            storage_key,
1724            &[(u64::MAX, vec![100, 200, 300])],
1725            &[
1726                // Before first entry -> NotYetWritten
1727                HistoryQuery {
1728                    block_number: 50,
1729                    lowest_available: None,
1730                    expected: HistoryInfo::NotYetWritten,
1731                },
1732                // Between entries -> InChangeset(next_write)
1733                HistoryQuery {
1734                    block_number: 150,
1735                    lowest_available: None,
1736                    expected: HistoryInfo::InChangeset(200),
1737                },
1738                // After last entry -> InPlainState
1739                HistoryQuery {
1740                    block_number: 500,
1741                    lowest_available: None,
1742                    expected: HistoryInfo::InPlainState,
1743                },
1744            ],
1745        );
1746
1747        // No history for different storage key
1748        run_storage_history_scenario(
1749            "storage_no_history",
1750            address,
1751            other_storage_key,
1752            &[], // No shards for this storage key
1753            &[HistoryQuery {
1754                block_number: 150,
1755                lowest_available: None,
1756                expected: HistoryInfo::NotYetWritten,
1757            }],
1758        );
1759    }
1760
1761    /// Test that `RocksDB` batches created via `EitherWriter` are only made visible when
1762    /// `provider.commit()` is called, not when the writer is dropped.
1763    #[test]
1764    fn test_rocksdb_commits_at_provider_level() {
1765        let factory = create_test_provider_factory();
1766
1767        // Enable RocksDB for transaction hash numbers
1768        factory.set_storage_settings_cache(StorageSettings::v2());
1769
1770        let hash1 = B256::from([1u8; 32]);
1771        let hash2 = B256::from([2u8; 32]);
1772        let tx_num1 = 100u64;
1773        let tx_num2 = 200u64;
1774
1775        // Get the RocksDB batch from the provider
1776        let rocksdb = factory.rocksdb_provider();
1777        let batch = rocksdb.batch();
1778
1779        // Create provider and EitherWriter
1780        let provider = factory.database_provider_rw().unwrap();
1781        let mut writer = EitherWriter::new_transaction_hash_numbers(&provider, batch).unwrap();
1782
1783        // Write transaction hash numbers (append_only=false since we're using RocksDB)
1784        writer.put_transaction_hash_number(hash1, tx_num1, false).unwrap();
1785        writer.put_transaction_hash_number(hash2, tx_num2, false).unwrap();
1786
1787        // Extract the raw batch from the writer and register it with the provider
1788        let raw_batch = writer.into_raw_rocksdb_batch();
1789        if let Some(batch) = raw_batch {
1790            provider.set_pending_rocksdb_batch(batch);
1791        }
1792
1793        // Data should NOT be visible yet (batch not committed)
1794        let rocksdb = factory.rocksdb_provider();
1795        assert_eq!(
1796            rocksdb.get::<tables::TransactionHashNumbers>(hash1).unwrap(),
1797            None,
1798            "Data should not be visible before provider.commit()"
1799        );
1800
1801        // Commit the provider - this should commit both MDBX and RocksDB
1802        provider.commit().unwrap();
1803
1804        // Now data should be visible in RocksDB
1805        let rocksdb = factory.rocksdb_provider();
1806        assert_eq!(
1807            rocksdb.get::<tables::TransactionHashNumbers>(hash1).unwrap(),
1808            Some(tx_num1),
1809            "Data should be visible after provider.commit()"
1810        );
1811        assert_eq!(
1812            rocksdb.get::<tables::TransactionHashNumbers>(hash2).unwrap(),
1813            Some(tx_num2),
1814            "Data should be visible after provider.commit()"
1815        );
1816    }
1817
1818    /// Test that `EitherReader::new_accounts_history` panics when settings require
1819    /// `RocksDB` but no snapshot is given (`None`). This is an invariant violation that
1820    /// indicates a bug - `with_rocksdb_snapshot` should always provide a snapshot when needed.
1821    #[test]
1822    #[should_panic(expected = "account_history_in_rocksdb requires rocksdb snapshot")]
1823    fn test_settings_mismatch_panics() {
1824        let factory = create_test_provider_factory();
1825
1826        factory.set_storage_settings_cache(StorageSettings::v2());
1827
1828        let provider = factory.database_provider_ro().unwrap();
1829        let _ = EitherReader::<(), ()>::new_accounts_history(&provider, None);
1830    }
1831}