1use 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
42type EitherReaderTy<'a, P, T> = EitherReader<
44 'a,
45 CursorTy<<P as DbTxProvider>::Tx, T>,
46 <P as NodePrimitivesProvider>::Primitives,
47>;
48
49type DupEitherReaderTy<'a, P, T> = EitherReader<
51 'a,
52 DupCursorTy<<P as DbTxProvider>::Tx, T>,
53 <P as NodePrimitivesProvider>::Primitives,
54>;
55
56type DupEitherWriterTy<'a, P, T> = EitherWriter<
58 'a,
59 DupCursorMutTy<<P as DbTxProvider>::Tx, T>,
60 <P as NodePrimitivesProvider>::Primitives,
61>;
62
63type EitherWriterTy<'a, P, T> = EitherWriter<
65 'a,
66 CursorMutTy<<P as DbTxProvider>::Tx, T>,
67 <P as NodePrimitivesProvider>::Primitives,
68>;
69
70pub type RocksBatchArg<'a> = crate::providers::rocksdb::RocksDBBatch<'a>;
72
73pub type RawRocksDBBatch = rocksdb::WriteBatchWithTransaction<true>;
75
76pub type RocksDBRefArg<'a> = Option<crate::providers::rocksdb::RocksReadSnapshot<'a>>;
81
82#[derive(Debug, Display)]
84pub enum EitherWriter<'a, CURSOR, N> {
85 Database(CURSOR),
87 StaticFile(StaticFileProviderRWRefMut<'a, N>),
89 RocksDB(RocksDBBatch<'a>),
91}
92
93impl<'a> EitherWriter<'a, (), ()> {
94 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 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 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 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 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 receipts_in_static_files && !prune_modes.receipts_log_filter.is_empty()
196 {
197 EitherWriterDestination::Database
198 } else {
199 EitherWriterDestination::StaticFile
200 }
201 }
202
203 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 pub fn append_account_changeset(
617 &mut self,
618 block_number: BlockNumber,
619 mut changeset: Vec<AccountBeforeTx>,
620 ) -> ProviderResult<()> {
621 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 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#[derive(Debug, Display)]
675pub enum EitherReader<'a, CURSOR, N> {
676 Database(CURSOR, PhantomData<&'a ()>),
678 StaticFile(StaticFileProvider<N>, PhantomData<&'a ()>),
680 RocksDB(crate::providers::rocksdb::RocksReadSnapshot<'a>),
682}
683
684impl<'a> EitherReader<'a, (), ()> {
685 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 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 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 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 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 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 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 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 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 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 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 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#[derive(Debug, EnumIs)]
975pub enum EitherWriterDestination {
976 Database,
978 StaticFile,
980 RocksDB,
982}
983
984impl EitherWriterDestination {
985 pub fn senders<P>(provider: &P) -> Self
987 where
988 P: StorageSettingsCache,
989 {
990 if provider.cached_storage_settings().storage_v2 {
992 Self::StaticFile
993 } else {
994 Self::Database
995 }
996 }
997
998 pub fn account_changesets<P>(provider: &P) -> Self
1000 where
1001 P: StorageSettingsCache,
1002 {
1003 if provider.cached_storage_settings().storage_v2 {
1005 Self::StaticFile
1006 } else {
1007 Self::Database
1008 }
1009 }
1010
1011 pub fn storage_changesets<P>(provider: &P) -> Self
1013 where
1014 P: StorageSettingsCache,
1015 {
1016 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 #[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 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 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]
1174 fn test_either_writer_transaction_hash_numbers_with_rocksdb() {
1175 let factory = create_test_provider_factory();
1176
1177 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 let rocksdb = factory.rocksdb_provider();
1187 let batch = rocksdb.batch();
1188
1189 let provider = factory.database_provider_rw().unwrap();
1191 let mut writer = EitherWriter::new_transaction_hash_numbers(&provider, batch).unwrap();
1192
1193 assert!(matches!(writer, EitherWriter::RocksDB(_)));
1195
1196 writer.put_transaction_hash_number(hash1, tx_num1, false).unwrap();
1198 writer.put_transaction_hash_number(hash2, tx_num2, false).unwrap();
1199
1200 if let Some(batch) = writer.into_raw_rocksdb_batch() {
1202 provider.set_pending_rocksdb_batch(batch);
1203 }
1204
1205 provider.commit().unwrap();
1207
1208 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]
1216 fn test_either_writer_delete_transaction_hash_number_with_rocksdb() {
1217 let factory = create_test_provider_factory();
1218
1219 factory.set_storage_settings_cache(StorageSettings::v2());
1221
1222 let hash = B256::from([1u8; 32]);
1223 let tx_num = 100u64;
1224
1225 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 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 if let Some(batch) = writer.into_raw_rocksdb_batch() {
1238 provider.set_pending_rocksdb_batch(batch);
1239 }
1240 provider.commit().unwrap();
1241
1242 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 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 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 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 let mut batch = provider.batch();
1283 batch.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
1284 batch.commit().unwrap();
1285
1286 let tx = provider.tx();
1288 let result = tx.get::<tables::StoragesHistory>(key).unwrap();
1289 assert_eq!(result, Some(value));
1290
1291 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 let mut batch = provider.batch();
1306 batch.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
1307 batch.commit().unwrap();
1308
1309 let tx = provider.tx();
1311 let result = tx.get::<tables::AccountsHistory>(key).unwrap();
1312 assert_eq!(result, Some(value));
1313
1314 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 provider.put::<tables::TransactionHashNumbers>(hash, &tx_num).unwrap();
1328 assert_eq!(provider.get::<tables::TransactionHashNumbers>(hash).unwrap(), Some(tx_num));
1329
1330 let mut batch = provider.batch();
1332 batch.delete::<tables::TransactionHashNumbers>(hash).unwrap();
1333 batch.commit().unwrap();
1334
1335 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 provider.put::<tables::StoragesHistory>(key.clone(), &value).unwrap();
1350 assert!(provider.get::<tables::StoragesHistory>(key.clone()).unwrap().is_some());
1351
1352 let mut batch = provider.batch();
1354 batch.delete::<tables::StoragesHistory>(key.clone()).unwrap();
1355 batch.commit().unwrap();
1356
1357 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 provider.put::<tables::AccountsHistory>(key.clone(), &value).unwrap();
1371 assert!(provider.get::<tables::AccountsHistory>(key.clone()).unwrap().is_some());
1372
1373 let mut batch = provider.batch();
1375 batch.delete::<tables::AccountsHistory>(key.clone()).unwrap();
1376 batch.commit().unwrap();
1377
1378 assert_eq!(provider.get::<tables::AccountsHistory>(key).unwrap(), None);
1380 }
1381
1382 struct HistoryQuery {
1389 block_number: BlockNumber,
1390 lowest_available: Option<BlockNumber>,
1391 expected: HistoryInfo,
1392 }
1393
1394 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 fn run_account_history_scenario(
1407 scenario_name: &str,
1408 address: Address,
1409 shards: &[(BlockNumber, Vec<BlockNumber>)], queries: &[HistoryQuery],
1411 ) {
1412 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 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 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 drop(mdbx_writer);
1435 mdbx_provider.commit().unwrap();
1436 if let EitherWriter::RocksDB(batch) = rocks_writer {
1437 batch.commit().unwrap();
1438 }
1439
1440 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 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 let rocks_result = rocks_snapshot
1457 .account_history_info(address, query.block_number, query.lowest_available, u64::MAX)
1458 .unwrap();
1459
1460 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 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 fn run_storage_history_scenario(
1495 scenario_name: &str,
1496 address: Address,
1497 storage_key: B256,
1498 shards: &[(BlockNumber, Vec<BlockNumber>)], queries: &[HistoryQuery],
1500 ) {
1501 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 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 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 drop(mdbx_writer);
1524 mdbx_provider.commit().unwrap();
1525 if let EitherWriter::RocksDB(batch) = rocks_writer {
1526 batch.commit().unwrap();
1527 }
1528
1529 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 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 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_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 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 #[test]
1601 fn test_account_history_info_both_backends() {
1602 let address = Address::from([0x42; 20]);
1603
1604 run_account_history_scenario(
1606 "single_shard",
1607 address,
1608 &[(u64::MAX, vec![100, 200, 300])],
1609 &[
1610 HistoryQuery {
1612 block_number: 50,
1613 lowest_available: None,
1614 expected: HistoryInfo::NotYetWritten,
1615 },
1616 HistoryQuery {
1618 block_number: 150,
1619 lowest_available: None,
1620 expected: HistoryInfo::InChangeset(200),
1621 },
1622 HistoryQuery {
1624 block_number: 300,
1625 lowest_available: None,
1626 expected: HistoryInfo::InChangeset(300),
1627 },
1628 HistoryQuery {
1630 block_number: 500,
1631 lowest_available: None,
1632 expected: HistoryInfo::InPlainState,
1633 },
1634 ],
1635 );
1636
1637 run_account_history_scenario(
1639 "multiple_shards",
1640 address,
1641 &[
1642 (500, vec![100, 200, 300, 400, 500]), (u64::MAX, vec![600, 700, 800]), ],
1645 &[
1646 HistoryQuery {
1648 block_number: 50,
1649 lowest_available: None,
1650 expected: HistoryInfo::NotYetWritten,
1651 },
1652 HistoryQuery {
1654 block_number: 150,
1655 lowest_available: None,
1656 expected: HistoryInfo::InChangeset(200),
1657 },
1658 HistoryQuery {
1660 block_number: 550,
1661 lowest_available: None,
1662 expected: HistoryInfo::InChangeset(600),
1663 },
1664 HistoryQuery {
1666 block_number: 900,
1667 lowest_available: None,
1668 expected: HistoryInfo::InPlainState,
1669 },
1670 ],
1671 );
1672
1673 let address_without_history = Address::from([0x43; 20]);
1675 run_account_history_scenario(
1676 "no_history",
1677 address_without_history,
1678 &[], &[HistoryQuery {
1680 block_number: 150,
1681 lowest_available: None,
1682 expected: HistoryInfo::NotYetWritten,
1683 }],
1684 );
1685
1686 run_account_history_scenario(
1692 "with_pruning_boundary",
1693 address,
1694 &[(u64::MAX, vec![100, 200, 300])],
1695 &[
1696 HistoryQuery {
1698 block_number: 100,
1699 lowest_available: Some(100),
1700 expected: HistoryInfo::InChangeset(100),
1701 },
1702 HistoryQuery {
1704 block_number: 150,
1705 lowest_available: Some(100),
1706 expected: HistoryInfo::InChangeset(200),
1707 },
1708 ],
1709 );
1710 }
1711
1712 #[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 run_storage_history_scenario(
1721 "storage_single_shard",
1722 address,
1723 storage_key,
1724 &[(u64::MAX, vec![100, 200, 300])],
1725 &[
1726 HistoryQuery {
1728 block_number: 50,
1729 lowest_available: None,
1730 expected: HistoryInfo::NotYetWritten,
1731 },
1732 HistoryQuery {
1734 block_number: 150,
1735 lowest_available: None,
1736 expected: HistoryInfo::InChangeset(200),
1737 },
1738 HistoryQuery {
1740 block_number: 500,
1741 lowest_available: None,
1742 expected: HistoryInfo::InPlainState,
1743 },
1744 ],
1745 );
1746
1747 run_storage_history_scenario(
1749 "storage_no_history",
1750 address,
1751 other_storage_key,
1752 &[], &[HistoryQuery {
1754 block_number: 150,
1755 lowest_available: None,
1756 expected: HistoryInfo::NotYetWritten,
1757 }],
1758 );
1759 }
1760
1761 #[test]
1764 fn test_rocksdb_commits_at_provider_level() {
1765 let factory = create_test_provider_factory();
1766
1767 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 let rocksdb = factory.rocksdb_provider();
1777 let batch = rocksdb.batch();
1778
1779 let provider = factory.database_provider_rw().unwrap();
1781 let mut writer = EitherWriter::new_transaction_hash_numbers(&provider, batch).unwrap();
1782
1783 writer.put_transaction_hash_number(hash1, tx_num1, false).unwrap();
1785 writer.put_transaction_hash_number(hash2, tx_num2, false).unwrap();
1786
1787 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 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 provider.commit().unwrap();
1803
1804 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]
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}