1use super::{
2 metrics::StaticFileProviderMetrics, writer::StaticFileWriters, LoadedJar,
3 StaticFileJarProvider, StaticFileProviderRW, StaticFileProviderRWRefMut,
4};
5use crate::{
6 changeset_walker::{StaticFileAccountChangesetWalker, StaticFileStorageChangesetWalker},
7 to_range, BlockHashReader, BlockNumReader, BlockReader, BlockSource, EitherWriter,
8 EitherWriterDestination, HeaderProvider, ReceiptProvider, StageCheckpointReader, StatsReader,
9 TransactionVariant, TransactionsProvider, TransactionsProviderExt,
10};
11use alloy_consensus::{
12 transaction::{TransactionMeta, TxHashRef},
13 Header,
14};
15use alloy_eips::BlockHashOrNumber;
16use alloy_primitives::{b256, Address, BlockHash, BlockNumber, TxHash, TxNumber, B256};
17
18use parking_lot::RwLock;
19use reth_chain_state::ExecutedBlock;
20use reth_chainspec::{ChainInfo, ChainSpecProvider, EthChainSpec, NamedChain};
21use reth_db::{
22 lockfile::StorageLock,
23 static_file::{
24 iter_static_files, BlockHashMask, HeaderMask, HeaderWithHashMask, ReceiptMask,
25 StaticFileCursor, StorageChangesetMask, TransactionMask, TransactionSenderMask,
26 },
27};
28use reth_db_api::{
29 cursor::DbCursorRO,
30 models::{AccountBeforeTx, BlockNumberAddress, StorageBeforeTx, StoredBlockBodyIndices},
31 table::{Decompress, Table, Value},
32 tables,
33 transaction::DbTx,
34};
35use reth_ethereum_primitives::{Receipt, TransactionSigned};
36use reth_nippy_jar::{NippyJar, NippyJarChecker};
37use reth_node_types::NodePrimitives;
38use reth_primitives_traits::{
39 dashmap::DashMap, AlloyBlockHeader as _, BlockBody as _, RecoveredBlock, SealedHeader,
40 SignedTransaction, StorageEntry,
41};
42use reth_prune_types::PruneSegment;
43use reth_stages_types::PipelineTarget;
44use reth_static_file_types::{
45 find_fixed_range, HighestStaticFiles, SegmentHeader, SegmentRangeInclusive, StaticFileMap,
46 StaticFileSegment, DEFAULT_BLOCKS_PER_STATIC_FILE,
47};
48use reth_storage_api::{
49 BlockBodyIndicesProvider, ChangeSetReader, DBProvider, PruneCheckpointReader,
50 StorageChangeSetReader, StorageSettingsCache,
51};
52use reth_storage_errors::provider::{ProviderError, ProviderResult, StaticFileWriterError};
53use std::{
54 collections::BTreeMap,
55 fmt::Debug,
56 ops::{Bound, Deref, Range, RangeBounds, RangeInclusive},
57 path::{Path, PathBuf},
58 sync::{atomic::AtomicU64, mpsc, Arc},
59};
60use tracing::{debug, info, info_span, instrument, trace, warn};
61
62type SegmentRanges = BTreeMap<u64, SegmentRangeInclusive>;
65
66#[derive(Debug, Default, PartialEq, Eq)]
68pub enum StaticFileAccess {
69 #[default]
71 RO,
72 RW,
74}
75
76impl StaticFileAccess {
77 pub const fn is_read_only(&self) -> bool {
79 matches!(self, Self::RO)
80 }
81
82 pub const fn is_read_write(&self) -> bool {
84 matches!(self, Self::RW)
85 }
86}
87
88#[derive(Debug, Clone, Copy, Default)]
92pub struct StaticFileWriteCtx {
93 pub write_senders: bool,
95 pub write_receipts: bool,
97 pub write_account_changesets: bool,
99 pub write_storage_changesets: bool,
101 pub tip: BlockNumber,
103 pub receipts_prune_mode: Option<reth_prune_types::PruneMode>,
105 pub receipts_prunable: bool,
107}
108
109#[derive(Debug)]
118pub struct StaticFileProvider<N>(pub(crate) Arc<StaticFileProviderInner<N>>);
119
120impl<N> Clone for StaticFileProvider<N> {
121 fn clone(&self) -> Self {
122 Self(self.0.clone())
123 }
124}
125
126#[derive(Debug)]
128pub struct StaticFileProviderBuilder<P> {
129 access: StaticFileAccess,
130 use_metrics: bool,
131 blocks_per_file: StaticFileMap<u64>,
132 path: P,
133 genesis_block_number: u64,
134}
135
136impl<P: AsRef<Path>> StaticFileProviderBuilder<P> {
137 pub fn read_write(path: P) -> Self {
139 Self {
140 path,
141 access: StaticFileAccess::RW,
142 blocks_per_file: Default::default(),
143 use_metrics: false,
144 genesis_block_number: 0,
145 }
146 }
147
148 pub fn read_only(path: P) -> Self {
150 Self {
151 path,
152 access: StaticFileAccess::RO,
153 blocks_per_file: Default::default(),
154 use_metrics: false,
155 genesis_block_number: 0,
156 }
157 }
158
159 pub fn with_blocks_per_file_for_segments(
171 mut self,
172 segments: &<StaticFileMap<u64> as Deref>::Target,
173 ) -> Self {
174 for (segment, &blocks_per_file) in segments {
175 self.blocks_per_file.insert(segment, blocks_per_file);
176 }
177 self
178 }
179
180 pub fn with_blocks_per_file(mut self, blocks_per_file: u64) -> Self {
182 for segment in StaticFileSegment::iter() {
183 self.blocks_per_file.insert(segment, blocks_per_file);
184 }
185 self
186 }
187
188 pub fn with_blocks_per_file_for_segment(
190 mut self,
191 segment: StaticFileSegment,
192 blocks_per_file: u64,
193 ) -> Self {
194 self.blocks_per_file.insert(segment, blocks_per_file);
195 self
196 }
197
198 pub const fn with_metrics(mut self) -> Self {
200 self.use_metrics = true;
201 self
202 }
203
204 pub const fn with_genesis_block_number(mut self, genesis_block_number: u64) -> Self {
217 self.genesis_block_number = genesis_block_number;
218 self
219 }
220
221 pub fn build<N: NodePrimitives>(self) -> ProviderResult<StaticFileProvider<N>> {
223 let mut provider = StaticFileProviderInner::new(self.path, self.access)?;
224 if self.use_metrics {
225 provider.metrics = Some(Arc::new(StaticFileProviderMetrics::default()));
226 }
227
228 for (segment, blocks_per_file) in *self.blocks_per_file {
229 provider.blocks_per_file.insert(segment, blocks_per_file);
230 }
231 provider.genesis_block_number = self.genesis_block_number;
232
233 let provider = StaticFileProvider(Arc::new(provider));
234 provider.initialize_index()?;
235 Ok(provider)
236 }
237}
238
239impl<N: NodePrimitives> StaticFileProvider<N> {
240 fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
242 let provider = Self(Arc::new(StaticFileProviderInner::new(path, access)?));
243 provider.initialize_index()?;
244 Ok(provider)
245 }
246}
247
248impl<N: NodePrimitives> StaticFileProvider<N> {
249 pub fn read_only(path: impl AsRef<Path>) -> ProviderResult<Self> {
254 Self::new(path, StaticFileAccess::RO)
255 }
256
257 pub fn read_write(path: impl AsRef<Path>) -> ProviderResult<Self> {
259 Self::new(path, StaticFileAccess::RW)
260 }
261}
262
263impl<N: NodePrimitives> Deref for StaticFileProvider<N> {
264 type Target = StaticFileProviderInner<N>;
265
266 fn deref(&self) -> &Self::Target {
267 &self.0
268 }
269}
270
271#[derive(Debug)]
273pub struct StaticFileProviderInner<N> {
274 map: DashMap<(BlockNumber, StaticFileSegment), LoadedJar>,
277 indexes: RwLock<StaticFileMap<StaticFileSegmentIndex>>,
279 earliest_history_height: AtomicU64,
290 path: PathBuf,
292 writers: StaticFileWriters<N>,
294 metrics: Option<Arc<StaticFileProviderMetrics>>,
296 access: StaticFileAccess,
298 blocks_per_file: StaticFileMap<u64>,
300 _lock_file: Option<StorageLock>,
302 genesis_block_number: u64,
304}
305
306impl<N: NodePrimitives> StaticFileProviderInner<N> {
307 fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
309 let _lock_file = if access.is_read_write() {
310 StorageLock::try_acquire(path.as_ref()).map_err(ProviderError::other)?.into()
311 } else {
312 None
313 };
314
315 let mut blocks_per_file = StaticFileMap::default();
316 for segment in StaticFileSegment::iter() {
317 blocks_per_file.insert(segment, DEFAULT_BLOCKS_PER_STATIC_FILE);
318 }
319
320 let provider = Self {
321 map: Default::default(),
322 indexes: Default::default(),
323 writers: Default::default(),
324 earliest_history_height: Default::default(),
325 path: path.as_ref().to_path_buf(),
326 metrics: None,
327 access,
328 blocks_per_file,
329 _lock_file,
330 genesis_block_number: 0,
331 };
332
333 Ok(provider)
334 }
335
336 pub const fn is_read_only(&self) -> bool {
337 self.access.is_read_only()
338 }
339
340 pub fn find_fixed_range_with_block_index(
349 &self,
350 segment: StaticFileSegment,
351 block_index: Option<&SegmentRanges>,
352 block: BlockNumber,
353 ) -> SegmentRangeInclusive {
354 let blocks_per_file =
355 self.blocks_per_file.get(segment).copied().unwrap_or(DEFAULT_BLOCKS_PER_STATIC_FILE);
356
357 if let Some(block_index) = block_index {
358 if let Some((_, range)) = block_index.range(block..).next() {
360 return *range;
362 } else if let Some((_, range)) = block_index.last_key_value() {
363 let blocks_after_last_range = block - range.end();
369 let segments_to_skip = (blocks_after_last_range - 1) / blocks_per_file;
370 let start = range.end() + 1 + segments_to_skip * blocks_per_file;
371 return SegmentRangeInclusive::new(start, start + blocks_per_file - 1);
372 }
373 }
374 find_fixed_range(block, blocks_per_file)
377 }
378
379 pub fn find_fixed_range(
392 &self,
393 segment: StaticFileSegment,
394 block: BlockNumber,
395 ) -> SegmentRangeInclusive {
396 self.find_fixed_range_with_block_index(
397 segment,
398 self.indexes.read().get(segment).map(|index| &index.expected_block_ranges_by_max_block),
399 block,
400 )
401 }
402
403 pub const fn genesis_block_number(&self) -> u64 {
405 self.genesis_block_number
406 }
407}
408
409impl<N: NodePrimitives> StaticFileProvider<N> {
410 pub fn report_metrics(&self) -> ProviderResult<()> {
415 let Some(metrics) = &self.metrics else { return Ok(()) };
416
417 let static_files = iter_static_files(&self.path).map_err(ProviderError::other)?;
418 for (segment, headers) in &*static_files {
419 let mut entries = 0;
420 let mut size = 0;
421
422 for (block_range, _) in headers {
423 let fixed_block_range = self.find_fixed_range(segment, block_range.start());
424 let jar_provider = self
425 .get_segment_provider_for_range(segment, || Some(fixed_block_range), None)?
426 .ok_or_else(|| {
427 ProviderError::MissingStaticFileBlock(segment, block_range.start())
428 })?;
429
430 entries += jar_provider.rows();
431 size += jar_provider.size() as u64;
432 }
433
434 metrics.record_segment(segment, size, headers.len(), entries);
435 }
436
437 Ok(())
438 }
439
440 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
442 fn write_headers(
443 w: &mut StaticFileProviderRWRefMut<'_, N>,
444 blocks: &[ExecutedBlock<N>],
445 ) -> ProviderResult<()> {
446 for block in blocks {
447 let b = block.recovered_block();
448 w.append_header(b.header(), &b.hash())?;
449 }
450 Ok(())
451 }
452
453 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
455 fn write_transactions(
456 w: &mut StaticFileProviderRWRefMut<'_, N>,
457 blocks: &[ExecutedBlock<N>],
458 tx_nums: &[TxNumber],
459 ) -> ProviderResult<()> {
460 for (block, &first_tx) in blocks.iter().zip(tx_nums) {
461 let b = block.recovered_block();
462 w.increment_block(b.number())?;
463 for (i, tx) in b.body().transactions().iter().enumerate() {
464 w.append_transaction(first_tx + i as u64, tx)?;
465 }
466 }
467 Ok(())
468 }
469
470 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
472 fn write_transaction_senders(
473 w: &mut StaticFileProviderRWRefMut<'_, N>,
474 blocks: &[ExecutedBlock<N>],
475 tx_nums: &[TxNumber],
476 ) -> ProviderResult<()> {
477 for (block, &first_tx) in blocks.iter().zip(tx_nums) {
478 let b = block.recovered_block();
479 w.increment_block(b.number())?;
480 for (i, sender) in b.senders_iter().enumerate() {
481 w.append_transaction_sender(first_tx + i as u64, sender)?;
482 }
483 }
484 Ok(())
485 }
486
487 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
489 fn write_receipts(
490 w: &mut StaticFileProviderRWRefMut<'_, N>,
491 blocks: &[ExecutedBlock<N>],
492 tx_nums: &[TxNumber],
493 ctx: &StaticFileWriteCtx,
494 ) -> ProviderResult<()> {
495 for (block, &first_tx) in blocks.iter().zip(tx_nums) {
496 let block_number = block.recovered_block().number();
497 w.increment_block(block_number)?;
498
499 if ctx.receipts_prunable &&
501 ctx.receipts_prune_mode
502 .is_some_and(|mode| mode.should_prune(block_number, ctx.tip))
503 {
504 continue
505 }
506
507 for (i, receipt) in block.execution_outcome().receipts.iter().enumerate() {
508 w.append_receipt(first_tx + i as u64, receipt)?;
509 }
510 }
511 Ok(())
512 }
513
514 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
516 fn write_account_changesets(
517 w: &mut StaticFileProviderRWRefMut<'_, N>,
518 blocks: &[ExecutedBlock<N>],
519 ) -> ProviderResult<()> {
520 for block in blocks {
521 let block_number = block.recovered_block().number();
522 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
523
524 let changeset: Vec<_> = reverts
525 .accounts
526 .into_iter()
527 .flatten()
528 .map(|(address, info)| AccountBeforeTx { address, info: info.map(Into::into) })
529 .collect();
530 w.append_account_changeset(changeset, block_number)?;
531 }
532 Ok(())
533 }
534
535 #[instrument(level = "debug", target = "providers::db", skip_all)]
537 fn write_storage_changesets(
538 w: &mut StaticFileProviderRWRefMut<'_, N>,
539 blocks: &[ExecutedBlock<N>],
540 ) -> ProviderResult<()> {
541 for block in blocks {
542 let block_number = block.recovered_block().number();
543 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
544
545 let changeset: Vec<_> = reverts
546 .storage
547 .into_iter()
548 .flatten()
549 .flat_map(|revert| {
550 revert.storage_revert.into_iter().map(move |(key, revert_to_slot)| {
551 StorageBeforeTx {
552 address: revert.address,
553 key: B256::from(key.to_be_bytes()),
554 value: revert_to_slot.to_previous_value(),
555 }
556 })
557 })
558 .collect();
559 w.append_storage_changeset(changeset, block_number)?;
560 }
561 Ok(())
562 }
563
564 #[instrument(level = "debug", target = "providers::static_file", skip_all, fields(?segment))]
569 fn write_segment<F>(
570 &self,
571 segment: StaticFileSegment,
572 first_block_number: BlockNumber,
573 f: F,
574 ) -> ProviderResult<()>
575 where
576 F: FnOnce(&mut StaticFileProviderRWRefMut<'_, N>) -> ProviderResult<()>,
577 {
578 let mut w = self.get_writer(first_block_number, segment)?;
579 f(&mut w)?;
580 w.sync_all()
581 }
582
583 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
588 pub fn write_blocks_data(
589 &self,
590 blocks: &[ExecutedBlock<N>],
591 tx_nums: &[TxNumber],
592 ctx: StaticFileWriteCtx,
593 runtime: &reth_tasks::Runtime,
594 ) -> ProviderResult<()> {
595 if blocks.is_empty() {
596 return Ok(());
597 }
598
599 let first_block_number = blocks[0].recovered_block().number();
600
601 let mut r_headers = None;
602 let mut r_txs = None;
603 let mut r_senders = None;
604 let mut r_receipts = None;
605 let mut r_account_changesets = None;
606 let mut r_storage_changesets = None;
607
608 let span = tracing::Span::current();
611 runtime.storage_pool().in_place_scope(|s| {
612 s.spawn(|_| {
613 let _guard = span.enter();
614 r_headers =
615 Some(self.write_segment(StaticFileSegment::Headers, first_block_number, |w| {
616 Self::write_headers(w, blocks)
617 }));
618 });
619
620 s.spawn(|_| {
621 let _guard = span.enter();
622 r_txs = Some(self.write_segment(
623 StaticFileSegment::Transactions,
624 first_block_number,
625 |w| Self::write_transactions(w, blocks, tx_nums),
626 ));
627 });
628
629 if ctx.write_senders {
630 s.spawn(|_| {
631 let _guard = span.enter();
632 r_senders = Some(self.write_segment(
633 StaticFileSegment::TransactionSenders,
634 first_block_number,
635 |w| Self::write_transaction_senders(w, blocks, tx_nums),
636 ));
637 });
638 }
639
640 if ctx.write_receipts {
641 s.spawn(|_| {
642 let _guard = span.enter();
643 r_receipts = Some(self.write_segment(
644 StaticFileSegment::Receipts,
645 first_block_number,
646 |w| Self::write_receipts(w, blocks, tx_nums, &ctx),
647 ));
648 });
649 }
650
651 if ctx.write_account_changesets {
652 s.spawn(|_| {
653 let _guard = span.enter();
654 r_account_changesets = Some(self.write_segment(
655 StaticFileSegment::AccountChangeSets,
656 first_block_number,
657 |w| Self::write_account_changesets(w, blocks),
658 ));
659 });
660 }
661
662 if ctx.write_storage_changesets {
663 s.spawn(|_| {
664 let _guard = span.enter();
665 r_storage_changesets = Some(self.write_segment(
666 StaticFileSegment::StorageChangeSets,
667 first_block_number,
668 |w| Self::write_storage_changesets(w, blocks),
669 ));
670 });
671 }
672 });
673
674 r_headers.ok_or(StaticFileWriterError::ThreadPanic("headers"))??;
675 r_txs.ok_or(StaticFileWriterError::ThreadPanic("transactions"))??;
676 if ctx.write_senders {
677 r_senders.ok_or(StaticFileWriterError::ThreadPanic("senders"))??;
678 }
679 if ctx.write_receipts {
680 r_receipts.ok_or(StaticFileWriterError::ThreadPanic("receipts"))??;
681 }
682 if ctx.write_account_changesets {
683 r_account_changesets
684 .ok_or(StaticFileWriterError::ThreadPanic("account_changesets"))??;
685 }
686 if ctx.write_storage_changesets {
687 r_storage_changesets
688 .ok_or(StaticFileWriterError::ThreadPanic("storage_changesets"))??;
689 }
690 Ok(())
691 }
692
693 pub fn get_segment_provider(
696 &self,
697 segment: StaticFileSegment,
698 number: u64,
699 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
700 if segment.is_block_or_change_based() {
701 self.get_segment_provider_for_block(segment, number, None)
702 } else {
703 self.get_segment_provider_for_transaction(segment, number, None)
704 }
705 }
706
707 pub fn get_maybe_segment_provider(
712 &self,
713 segment: StaticFileSegment,
714 number: u64,
715 ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
716 let provider = if segment.is_block_or_change_based() {
717 self.get_segment_provider_for_block(segment, number, None)
718 } else {
719 self.get_segment_provider_for_transaction(segment, number, None)
720 };
721
722 match provider {
723 Ok(provider) => Ok(Some(provider)),
724 Err(
725 ProviderError::MissingStaticFileBlock(_, _) |
726 ProviderError::MissingStaticFileTx(_, _),
727 ) => Ok(None),
728 Err(err) => Err(err),
729 }
730 }
731
732 pub fn get_segment_provider_for_block(
734 &self,
735 segment: StaticFileSegment,
736 block: BlockNumber,
737 path: Option<&Path>,
738 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
739 self.get_segment_provider_for_range(
740 segment,
741 || self.get_segment_ranges_from_block(segment, block),
742 path,
743 )?
744 .ok_or(ProviderError::MissingStaticFileBlock(segment, block))
745 }
746
747 pub fn get_segment_provider_for_transaction(
749 &self,
750 segment: StaticFileSegment,
751 tx: TxNumber,
752 path: Option<&Path>,
753 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
754 self.get_segment_provider_for_range(
755 segment,
756 || self.get_segment_ranges_from_transaction(segment, tx),
757 path,
758 )?
759 .ok_or(ProviderError::MissingStaticFileTx(segment, tx))
760 }
761
762 pub fn get_segment_provider_for_range(
766 &self,
767 segment: StaticFileSegment,
768 fn_range: impl Fn() -> Option<SegmentRangeInclusive>,
769 path: Option<&Path>,
770 ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
771 let block_range = match path {
774 Some(path) => StaticFileSegment::parse_filename(
775 &path
776 .file_name()
777 .ok_or_else(|| {
778 ProviderError::MissingStaticFileSegmentPath(segment, path.to_path_buf())
779 })?
780 .to_string_lossy(),
781 )
782 .and_then(|(parsed_segment, block_range)| {
783 if parsed_segment == segment {
784 return Some(block_range);
785 }
786 None
787 }),
788 None => fn_range(),
789 };
790
791 if let Some(block_range) = block_range {
793 return Ok(Some(self.get_or_create_jar_provider(segment, &block_range)?));
794 }
795
796 Ok(None)
797 }
798
799 pub fn get_segment_provider_for_path(
801 &self,
802 path: &Path,
803 ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
804 StaticFileSegment::parse_filename(
805 &path
806 .file_name()
807 .ok_or_else(|| ProviderError::MissingStaticFilePath(path.to_path_buf()))?
808 .to_string_lossy(),
809 )
810 .map(|(segment, block_range)| self.get_or_create_jar_provider(segment, &block_range))
811 .transpose()
812 }
813
814 pub fn remove_cached_provider(
818 &self,
819 segment: StaticFileSegment,
820 fixed_block_range_end: BlockNumber,
821 ) {
822 self.map.remove(&(fixed_block_range_end, segment));
823 }
824
825 pub fn delete_segment_below_block(
842 &self,
843 segment: StaticFileSegment,
844 block: BlockNumber,
845 ) -> ProviderResult<Vec<SegmentHeader>> {
846 if block == 0 {
848 return Ok(Vec::new());
849 }
850
851 let highest_block = self.get_highest_static_file_block(segment);
852 let mut deleted_headers = Vec::new();
853
854 loop {
855 let Some(block_height) = self.get_lowest_range_end(segment) else {
856 return Ok(deleted_headers);
857 };
858
859 if block_height >= block || Some(block_height) == highest_block {
861 return Ok(deleted_headers);
862 }
863
864 debug!(
865 target: "providers::static_file",
866 ?segment,
867 ?block_height,
868 "Deleting static file below block"
869 );
870
871 let header = self.delete_jar(segment, block_height).inspect_err(|err| {
874 warn!( target: "providers::static_file", ?segment, %block_height, ?err, "Failed to delete static file below block")
875 })?;
876
877 deleted_headers.push(header);
878 }
879 }
880
881 pub fn delete_jar(
889 &self,
890 segment: StaticFileSegment,
891 block: BlockNumber,
892 ) -> ProviderResult<SegmentHeader> {
893 let fixed_block_range = self.find_fixed_range(segment, block);
894 let key = (fixed_block_range.end(), segment);
895 let file = self.path.join(segment.filename(&fixed_block_range));
896 let jar = if let Some((_, jar)) = self.map.remove(&key) {
897 jar.jar
898 } else {
899 debug!(
900 target: "providers::static_file",
901 ?file,
902 ?fixed_block_range,
903 ?block,
904 "Loading static file jar for deletion"
905 );
906 NippyJar::<SegmentHeader>::load(&file).map_err(ProviderError::other)?
907 };
908
909 let header = jar.user_header().clone();
910
911 if segment.is_change_based() {
913 let csoff_path = file.with_extension("csoff");
914 if csoff_path.exists() {
915 std::fs::remove_file(&csoff_path).map_err(ProviderError::other)?;
916 }
917 }
918
919 jar.delete().map_err(ProviderError::other)?;
920
921 self.initialize_index()?;
924
925 Ok(header)
926 }
927
928 pub fn delete_segment(&self, segment: StaticFileSegment) -> ProviderResult<Vec<SegmentHeader>> {
936 let mut deleted_headers = Vec::new();
937
938 self.writers.remove(segment);
939
940 while let Some(block_height) = self.get_highest_static_file_block(segment) {
941 debug!(
942 target: "providers::static_file",
943 ?segment,
944 ?block_height,
945 "Deleting static file jar"
946 );
947
948 let header = self.delete_jar(segment, block_height).inspect_err(|err| {
949 warn!(target: "providers::static_file", ?segment, %block_height, ?err, "Failed to delete static file jar")
950 })?;
951
952 deleted_headers.push(header);
953 }
954
955 Ok(deleted_headers)
956 }
957
958 fn get_or_create_jar_provider(
962 &self,
963 segment: StaticFileSegment,
964 fixed_block_range: &SegmentRangeInclusive,
965 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
966 let key = (fixed_block_range.end(), segment);
967
968 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Getting provider");
970 let mut provider: StaticFileJarProvider<'_, N> = if let Some(jar) = self.map.get(&key) {
971 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Jar found in cache");
972 jar.into()
973 } else {
974 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Creating jar from scratch");
975 let path = self.path.join(segment.filename(fixed_block_range));
976 let jar = NippyJar::load(&path).map_err(ProviderError::other)?;
977 let loaded = LoadedJar::new(jar)?;
978 self.map.entry(key).or_insert(loaded).downgrade().into()
982 };
983
984 if let Some(metrics) = &self.metrics {
985 provider = provider.with_metrics(metrics.clone());
986 }
987 Ok(provider)
988 }
989
990 fn get_segment_ranges_from_block(
993 &self,
994 segment: StaticFileSegment,
995 block: u64,
996 ) -> Option<SegmentRangeInclusive> {
997 let indexes = self.indexes.read();
998 let index = indexes.get(segment)?;
999
1000 (index.max_block >= block).then(|| {
1001 self.find_fixed_range_with_block_index(
1002 segment,
1003 Some(&index.expected_block_ranges_by_max_block),
1004 block,
1005 )
1006 })
1007 }
1008
1009 fn get_segment_ranges_from_transaction(
1012 &self,
1013 segment: StaticFileSegment,
1014 tx: u64,
1015 ) -> Option<SegmentRangeInclusive> {
1016 let indexes = self.indexes.read();
1017 let index = indexes.get(segment)?;
1018 let available_block_ranges_by_max_tx = index.available_block_ranges_by_max_tx.as_ref()?;
1019
1020 let mut static_files_rev_iter = available_block_ranges_by_max_tx.iter().rev().peekable();
1023
1024 while let Some((tx_end, block_range)) = static_files_rev_iter.next() {
1025 if tx > *tx_end {
1026 return None;
1028 }
1029 let tx_start = static_files_rev_iter.peek().map(|(tx_end, _)| *tx_end + 1).unwrap_or(0);
1030 if tx_start <= tx {
1031 return Some(self.find_fixed_range_with_block_index(
1032 segment,
1033 Some(&index.expected_block_ranges_by_max_block),
1034 block_range.end(),
1035 ));
1036 }
1037 }
1038 None
1039 }
1040
1041 pub fn update_index(
1048 &self,
1049 segment: StaticFileSegment,
1050 segment_max_block: Option<BlockNumber>,
1051 ) -> ProviderResult<()> {
1052 trace!(
1053 target: "providers::static_file",
1054 ?segment,
1055 ?segment_max_block,
1056 "Updating provider index"
1057 );
1058 let mut indexes = self.indexes.write();
1059
1060 match segment_max_block {
1061 Some(segment_max_block) => {
1062 let fixed_range = self.find_fixed_range_with_block_index(
1063 segment,
1064 indexes.get(segment).map(|index| &index.expected_block_ranges_by_max_block),
1065 segment_max_block,
1066 );
1067
1068 let jar = NippyJar::<SegmentHeader>::load(
1069 &self.path.join(segment.filename(&fixed_range)),
1070 )
1071 .map_err(ProviderError::other)?;
1072
1073 let index = indexes
1074 .entry(segment)
1075 .and_modify(|index| {
1076 index.max_block = segment_max_block;
1078
1079 index
1083 .expected_block_ranges_by_max_block
1084 .retain(|_, block_range| block_range.start() < fixed_range.start());
1085 index
1087 .expected_block_ranges_by_max_block
1088 .insert(fixed_range.end(), fixed_range);
1089 })
1090 .or_insert_with(|| StaticFileSegmentIndex {
1091 min_block_range: None,
1092 max_block: segment_max_block,
1093 expected_block_ranges_by_max_block: BTreeMap::from([(
1094 fixed_range.end(),
1095 fixed_range,
1096 )]),
1097 available_block_ranges_by_max_tx: None,
1098 });
1099
1100 if let Some(current_block_range) = jar.user_header().block_range() {
1116 if let Some(min_block_range) = index.min_block_range.as_mut() {
1117 if current_block_range.start() == min_block_range.start() {
1120 *min_block_range = current_block_range;
1121 }
1122 } else {
1123 index.min_block_range = Some(current_block_range);
1124 }
1125 }
1126
1127 if let Some(tx_range) = jar.user_header().tx_range() {
1130 if let Some(current_block_range) = jar.user_header().block_range() {
1133 let tx_end = tx_range.end();
1134
1135 if let Some(index) = index.available_block_ranges_by_max_tx.as_mut() {
1144 index
1145 .retain(|_, block_range| block_range.start() < fixed_range.start());
1146 index.insert(tx_end, current_block_range);
1147 } else {
1148 index.available_block_ranges_by_max_tx =
1149 Some(BTreeMap::from([(tx_end, current_block_range)]));
1150 }
1151 }
1152 } else if segment.is_tx_based() {
1153 if let Some(index) = index.available_block_ranges_by_max_tx.as_mut() {
1157 index.retain(|_, block_range| block_range.start() < fixed_range.start());
1158 }
1159
1160 index.available_block_ranges_by_max_tx.take_if(|index| index.is_empty());
1162 }
1163
1164 trace!(target: "providers::static_file", ?segment, "Inserting updated jar into cache");
1166 self.map.insert((fixed_range.end(), segment), LoadedJar::new(jar)?);
1167
1168 trace!(target: "providers::static_file", ?segment, "Cleaning up jar map");
1170 self.map.retain(|(end, seg), _| !(*seg == segment && *end > fixed_range.end()));
1171 }
1172 None => {
1173 debug!(target: "providers::static_file", ?segment, "Removing segment from index");
1174 indexes.remove(segment);
1175 }
1176 };
1177
1178 trace!(target: "providers::static_file", ?segment, "Updated provider index");
1179 Ok(())
1180 }
1181
1182 pub fn initialize_index(&self) -> ProviderResult<()> {
1184 let mut indexes = self.indexes.write();
1185 indexes.clear();
1186
1187 for (segment, headers) in &*iter_static_files(&self.path).map_err(ProviderError::other)? {
1188 let min_block_range = Some(headers.first().expect("headers are not empty").0);
1193 let max_block = headers.last().expect("headers are not empty").0.end();
1194
1195 let mut expected_block_ranges_by_max_block = BTreeMap::default();
1196 let mut available_block_ranges_by_max_tx = None;
1197
1198 for (block_range, header) in headers {
1199 expected_block_ranges_by_max_block
1201 .insert(header.expected_block_end(), header.expected_block_range());
1202
1203 if let Some(tx_range) = header.tx_range() {
1205 let tx_end = tx_range.end();
1206
1207 available_block_ranges_by_max_tx
1208 .get_or_insert_with(BTreeMap::default)
1209 .insert(tx_end, *block_range);
1210 }
1211 }
1212
1213 indexes.insert(
1214 segment,
1215 StaticFileSegmentIndex {
1216 min_block_range,
1217 max_block,
1218 expected_block_ranges_by_max_block,
1219 available_block_ranges_by_max_tx,
1220 },
1221 );
1222 }
1223
1224 self.map.clear();
1226
1227 if let Some(lowest_range) =
1229 indexes.get(StaticFileSegment::Transactions).and_then(|index| index.min_block_range)
1230 {
1231 self.earliest_history_height
1233 .store(lowest_range.start(), std::sync::atomic::Ordering::Relaxed);
1234 }
1235
1236 Ok(())
1237 }
1238
1239 #[instrument(skip(self, provider), fields(read_only = self.is_read_only()))]
1263 pub fn check_consistency<Provider>(
1264 &self,
1265 provider: &Provider,
1266 ) -> ProviderResult<Option<PipelineTarget>>
1267 where
1268 Provider: DBProvider
1269 + BlockReader
1270 + StageCheckpointReader
1271 + PruneCheckpointReader
1272 + ChainSpecProvider
1273 + StorageSettingsCache,
1274 N: NodePrimitives<Receipt: Value, BlockHeader: Value, SignedTx: Value>,
1275 {
1276 if provider.chain_spec().is_optimism() &&
1283 reth_chainspec::Chain::optimism_mainnet() == provider.chain_spec().chain_id()
1284 {
1285 const OVM_HEADER_1_HASH: B256 =
1287 b256!("0xbee7192e575af30420cae0c7776304ac196077ee72b048970549e4f08e875453");
1288 if provider.block_number(OVM_HEADER_1_HASH)?.is_some() {
1289 info!(target: "reth::cli",
1290 "Skipping storage verification for OP mainnet, expected inconsistency in OVM chain"
1291 );
1292 return Ok(None);
1293 }
1294 }
1295
1296 info!(target: "reth::cli", "Verifying storage consistency.");
1297
1298 let mut unwind_target: Option<BlockNumber> = None;
1299
1300 let mut update_unwind_target = |new_target| {
1301 unwind_target =
1302 unwind_target.map(|current| current.min(new_target)).or(Some(new_target));
1303 };
1304
1305 for segment in self.segments_to_check(provider) {
1306 let span = info_span!(
1307 "Checking consistency for segment",
1308 ?segment,
1309 initial_highest_block = tracing::field::Empty,
1310 highest_block = tracing::field::Empty,
1311 highest_tx = tracing::field::Empty,
1312 );
1313 let _guard = span.enter();
1314
1315 debug!(target: "reth::providers::static_file", "Checking consistency for segment");
1316
1317 let (initial_highest_block, mut highest_block) = self.maybe_heal_segment(segment)?;
1319 span.record("initial_highest_block", initial_highest_block);
1320 span.record("highest_block", highest_block);
1321
1322 if initial_highest_block != highest_block {
1327 info!(
1328 target: "reth::providers::static_file",
1329 unwind_target = highest_block,
1330 "Setting unwind target."
1331 );
1332 update_unwind_target(highest_block.unwrap_or_default());
1333 }
1334
1335 let highest_tx = self.get_highest_static_file_tx(segment);
1341 span.record("highest_tx", highest_tx);
1342 debug!(target: "reth::providers::static_file", "Checking tx index segment");
1343
1344 if let Some(highest_tx) = highest_tx {
1345 let mut last_block = highest_block.unwrap_or_default();
1346 debug!(target: "reth::providers::static_file", last_block, highest_tx, "Verifying last transaction matches last block indices");
1347 loop {
1348 let Some(indices) = provider.block_body_indices(last_block)? else {
1349 debug!(target: "reth::providers::static_file", last_block, "Block body indices not found, static files ahead of database");
1350 break
1354 };
1355
1356 debug!(target: "reth::providers::static_file", last_block, last_tx_num = indices.last_tx_num(), "Found block body indices");
1357
1358 if indices.last_tx_num() <= highest_tx {
1359 break
1360 }
1361
1362 if last_block == 0 {
1363 debug!(target: "reth::providers::static_file", "Reached block 0 in verification loop");
1364 break
1365 }
1366
1367 last_block -= 1;
1368
1369 info!(
1370 target: "reth::providers::static_file",
1371 highest_block = self.get_highest_static_file_block(segment),
1372 unwind_target = last_block,
1373 "Setting unwind target."
1374 );
1375 span.record("highest_block", last_block);
1376 highest_block = Some(last_block);
1377 update_unwind_target(last_block);
1378 }
1379 }
1380
1381 debug!(target: "reth::providers::static_file", "Ensuring invariants for segment");
1382
1383 match self.ensure_invariants_for(provider, segment, highest_tx, highest_block)? {
1384 Some(unwind) => {
1385 debug!(target: "reth::providers::static_file", unwind_target=unwind, "Invariants check returned unwind target");
1386 update_unwind_target(unwind);
1387 }
1388 None => {
1389 debug!(target: "reth::providers::static_file", "Invariants check completed, no unwind needed")
1390 }
1391 }
1392 }
1393
1394 Ok(unwind_target.map(PipelineTarget::Unwind))
1395 }
1396
1397 pub fn check_file_consistency<Provider>(&self, provider: &Provider) -> ProviderResult<()>
1406 where
1407 Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1408 {
1409 info!(target: "reth::cli", "Healing static file inconsistencies.");
1410
1411 for segment in self.segments_to_check(provider) {
1412 let _guard = info_span!("healing_static_file_segment", ?segment).entered();
1413 let _ = self.maybe_heal_segment(segment)?;
1414 }
1415
1416 Ok(())
1417 }
1418
1419 fn segments_to_check<'a, Provider>(
1421 &'a self,
1422 provider: &'a Provider,
1423 ) -> impl Iterator<Item = StaticFileSegment> + 'a
1424 where
1425 Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1426 {
1427 StaticFileSegment::iter()
1428 .filter(move |segment| self.should_check_segment(provider, *segment))
1429 }
1430
1431 fn should_check_segment<Provider>(
1433 &self,
1434 provider: &Provider,
1435 segment: StaticFileSegment,
1436 ) -> bool
1437 where
1438 Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1439 {
1440 match segment {
1441 StaticFileSegment::Headers | StaticFileSegment::Transactions => true,
1442 StaticFileSegment::Receipts => {
1443 if EitherWriter::receipts_destination(provider).is_database() {
1444 debug!(target: "reth::providers::static_file", ?segment, "Skipping receipts segment: receipts stored in database");
1447 return false;
1448 }
1449
1450 if NamedChain::Gnosis == provider.chain_spec().chain_id() ||
1451 NamedChain::Chiado == provider.chain_spec().chain_id()
1452 {
1453 debug!(target: "reth::providers::static_file", ?segment, "Skipping receipts segment: broken historical import for gnosis/chiado");
1457 return false;
1458 }
1459
1460 true
1461 }
1462 StaticFileSegment::TransactionSenders => {
1463 if EitherWriterDestination::senders(provider).is_database() {
1464 debug!(target: "reth::providers::static_file", ?segment, "Skipping senders segment: senders stored in database");
1465 return false;
1466 }
1467
1468 if Self::is_segment_fully_pruned(provider, PruneSegment::SenderRecovery) {
1469 debug!(target: "reth::providers::static_file", ?segment, "Skipping senders segment: fully pruned");
1470 return false;
1471 }
1472
1473 true
1474 }
1475 StaticFileSegment::AccountChangeSets => {
1476 if EitherWriter::account_changesets_destination(provider).is_database() {
1477 debug!(target: "reth::providers::static_file", ?segment, "Skipping account changesets segment: changesets stored in database");
1478 return false;
1479 }
1480 true
1481 }
1482 StaticFileSegment::StorageChangeSets => {
1483 if EitherWriter::storage_changesets_destination(provider).is_database() {
1484 debug!(target: "reth::providers::static_file", ?segment, "Skipping storage changesets segment: changesets stored in database");
1485 return false
1486 }
1487 true
1488 }
1489 }
1490 }
1491
1492 fn is_segment_fully_pruned<Provider>(provider: &Provider, segment: PruneSegment) -> bool
1496 where
1497 Provider: PruneCheckpointReader,
1498 {
1499 provider
1500 .get_prune_checkpoint(segment)
1501 .ok()
1502 .flatten()
1503 .is_some_and(|checkpoint| checkpoint.prune_mode.is_full())
1504 }
1505
1506 fn check_segment_consistency(&self, segment: StaticFileSegment) -> ProviderResult<()> {
1511 debug!(target: "reth::providers::static_file", "Checking segment consistency");
1512 if let Some(latest_block) = self.get_highest_static_file_block(segment) {
1513 let file_path = self
1514 .directory()
1515 .join(segment.filename(&self.find_fixed_range(segment, latest_block)));
1516 debug!(target: "reth::providers::static_file", ?file_path, latest_block, "Loading NippyJar for consistency check");
1517
1518 let jar = NippyJar::<SegmentHeader>::load(&file_path).map_err(ProviderError::other)?;
1519 debug!(target: "reth::providers::static_file", "NippyJar loaded, checking consistency");
1520
1521 NippyJarChecker::new(jar).check_consistency().map_err(ProviderError::other)?;
1522 debug!(target: "reth::providers::static_file", "NippyJar consistency check passed");
1523 } else {
1524 debug!(target: "reth::providers::static_file", "No static file block found, skipping consistency check");
1525 }
1526 Ok(())
1527 }
1528
1529 fn maybe_heal_segment(
1545 &self,
1546 segment: StaticFileSegment,
1547 ) -> ProviderResult<(Option<BlockNumber>, Option<BlockNumber>)> {
1548 let initial_highest_block = self.get_highest_static_file_block(segment);
1549 debug!(target: "reth::providers::static_file", ?initial_highest_block, "Initial highest block for segment");
1550
1551 if self.access.is_read_only() {
1552 debug!(target: "reth::providers::static_file", "Checking segment consistency (read-only)");
1555 self.check_segment_consistency(segment)?;
1556 } else {
1557 debug!(target: "reth::providers::static_file", "Fetching latest writer which might heal any potential inconsistency");
1560 self.latest_writer(segment)?;
1561 }
1562
1563 let highest_block = self.get_highest_static_file_block(segment);
1566
1567 Ok((initial_highest_block, highest_block))
1568 }
1569
1570 fn ensure_invariants_for<Provider>(
1572 &self,
1573 provider: &Provider,
1574 segment: StaticFileSegment,
1575 highest_tx: Option<u64>,
1576 highest_block: Option<BlockNumber>,
1577 ) -> ProviderResult<Option<BlockNumber>>
1578 where
1579 Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1580 N: NodePrimitives<Receipt: Value, BlockHeader: Value, SignedTx: Value>,
1581 {
1582 match segment {
1583 StaticFileSegment::Headers => self
1584 .ensure_invariants::<_, tables::Headers<N::BlockHeader>>(
1585 provider,
1586 segment,
1587 highest_block,
1588 highest_block,
1589 ),
1590 StaticFileSegment::Transactions => self
1591 .ensure_invariants::<_, tables::Transactions<N::SignedTx>>(
1592 provider,
1593 segment,
1594 highest_tx,
1595 highest_block,
1596 ),
1597 StaticFileSegment::Receipts => self
1598 .ensure_invariants::<_, tables::Receipts<N::Receipt>>(
1599 provider,
1600 segment,
1601 highest_tx,
1602 highest_block,
1603 ),
1604 StaticFileSegment::TransactionSenders => self
1605 .ensure_invariants::<_, tables::TransactionSenders>(
1606 provider,
1607 segment,
1608 highest_tx,
1609 highest_block,
1610 ),
1611 StaticFileSegment::AccountChangeSets => self
1612 .ensure_invariants::<_, tables::AccountChangeSets>(
1613 provider,
1614 segment,
1615 highest_tx,
1616 highest_block,
1617 ),
1618 StaticFileSegment::StorageChangeSets => self
1619 .ensure_changeset_invariants_by_block::<_, tables::StorageChangeSets, _>(
1620 provider,
1621 segment,
1622 highest_block,
1623 |key| key.block_number(),
1624 ),
1625 }
1626 }
1627
1628 #[instrument(skip(self, provider, segment), fields(table = T::NAME))]
1643 fn ensure_invariants<Provider, T: Table<Key = u64>>(
1644 &self,
1645 provider: &Provider,
1646 segment: StaticFileSegment,
1647 highest_static_file_entry: Option<u64>,
1648 highest_static_file_block: Option<BlockNumber>,
1649 ) -> ProviderResult<Option<BlockNumber>>
1650 where
1651 Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1652 {
1653 debug!(target: "reth::providers::static_file", "Ensuring invariants");
1654 let mut db_cursor = provider.tx_ref().cursor_read::<T>()?;
1655
1656 if let Some((db_first_entry, _)) = db_cursor.first()? {
1657 debug!(target: "reth::providers::static_file", db_first_entry, "Found first database entry");
1658 if let (Some(highest_entry), Some(highest_block)) =
1659 (highest_static_file_entry, highest_static_file_block)
1660 {
1661 if !(db_first_entry <= highest_entry || highest_entry + 1 == db_first_entry) {
1665 info!(
1666 target: "reth::providers::static_file",
1667 ?db_first_entry,
1668 ?highest_entry,
1669 unwind_target = highest_block,
1670 "Setting unwind target."
1671 );
1672 return Ok(Some(highest_block));
1673 }
1674 }
1675
1676 if let Some((db_last_entry, _)) = db_cursor.last()? &&
1677 highest_static_file_entry
1678 .is_none_or(|highest_entry| db_last_entry > highest_entry)
1679 {
1680 debug!(target: "reth::providers::static_file", db_last_entry, "Database has entries beyond static files, no unwind needed");
1681 return Ok(None)
1682 }
1683 } else {
1684 debug!(target: "reth::providers::static_file", "No database entries found");
1685 }
1686
1687 let highest_static_file_entry = highest_static_file_entry.unwrap_or_default();
1688 let highest_static_file_block = highest_static_file_block.unwrap_or_default();
1689
1690 let stage_id = segment.to_stage_id();
1693 let checkpoint_block_number =
1694 provider.get_stage_checkpoint(stage_id)?.unwrap_or_default().block_number;
1695 debug!(target: "reth::providers::static_file", ?stage_id, checkpoint_block_number, "Retrieved stage checkpoint");
1696
1697 let effective_coverage_block =
1698 Self::effective_coverage_block(provider, segment, highest_static_file_block)?;
1699
1700 if checkpoint_block_number > effective_coverage_block {
1702 info!(
1703 target: "reth::providers::static_file",
1704 checkpoint_block_number,
1705 unwind_target = effective_coverage_block,
1706 "Setting unwind target."
1707 );
1708 return Ok(Some(effective_coverage_block));
1709 }
1710
1711 if checkpoint_block_number >= highest_static_file_block {
1713 debug!(target: "reth::providers::static_file", "Invariants ensured, returning None");
1714 return Ok(None);
1715 }
1716
1717 info!(
1723 target: "reth::providers",
1724 from = highest_static_file_block,
1725 to = checkpoint_block_number,
1726 "Unwinding static file segment."
1727 );
1728 let mut writer = self.latest_writer(segment)?;
1729
1730 match segment {
1731 StaticFileSegment::Headers => {
1732 let prune_count = highest_static_file_block - checkpoint_block_number;
1733 debug!(target: "reth::providers::static_file", prune_count, "Pruning headers");
1734 writer.prune_headers(prune_count)?;
1736 }
1737 StaticFileSegment::Transactions |
1738 StaticFileSegment::Receipts |
1739 StaticFileSegment::TransactionSenders => {
1740 if let Some(block) = provider.block_body_indices(checkpoint_block_number)? {
1741 let number = highest_static_file_entry - block.last_tx_num();
1742 debug!(target: "reth::providers::static_file", prune_count = number, checkpoint_block_number, "Pruning transaction based segment");
1743
1744 match segment {
1745 StaticFileSegment::Transactions => {
1746 writer.prune_transactions(number, checkpoint_block_number)?
1747 }
1748 StaticFileSegment::Receipts => {
1749 writer.prune_receipts(number, checkpoint_block_number)?
1750 }
1751 StaticFileSegment::TransactionSenders => {
1752 writer.prune_transaction_senders(number, checkpoint_block_number)?
1753 }
1754 StaticFileSegment::Headers |
1755 StaticFileSegment::AccountChangeSets |
1756 StaticFileSegment::StorageChangeSets => {
1757 unreachable!()
1758 }
1759 }
1760 } else {
1761 debug!(target: "reth::providers::static_file", checkpoint_block_number, "No block body indices found for checkpoint block");
1762 }
1763 }
1764 StaticFileSegment::AccountChangeSets => {
1765 writer.prune_account_changesets(checkpoint_block_number)?;
1766 }
1767 StaticFileSegment::StorageChangeSets => {
1768 writer.prune_storage_changesets(checkpoint_block_number)?;
1769 }
1770 }
1771
1772 debug!(target: "reth::providers::static_file", "Committing writer after pruning");
1773 writer.commit()?;
1774 debug!(target: "reth::providers::static_file", "Writer committed successfully");
1775
1776 debug!(target: "reth::providers::static_file", "Invariants ensured, returning None");
1777 Ok(None)
1778 }
1779
1780 fn ensure_changeset_invariants_by_block<Provider, T, F>(
1781 &self,
1782 provider: &Provider,
1783 segment: StaticFileSegment,
1784 highest_static_file_block: Option<BlockNumber>,
1785 block_from_key: F,
1786 ) -> ProviderResult<Option<BlockNumber>>
1787 where
1788 Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1789 T: Table,
1790 F: Fn(&T::Key) -> BlockNumber,
1791 {
1792 debug!(
1793 target: "reth::providers::static_file",
1794 ?segment,
1795 ?highest_static_file_block,
1796 "Ensuring changeset invariants"
1797 );
1798 let mut db_cursor = provider.tx_ref().cursor_read::<T>()?;
1799
1800 if let Some((db_first_key, _)) = db_cursor.first()? {
1801 let db_first_block = block_from_key(&db_first_key);
1802 if let Some(highest_block) = highest_static_file_block &&
1803 !(db_first_block <= highest_block || highest_block + 1 == db_first_block)
1804 {
1805 info!(
1806 target: "reth::providers::static_file",
1807 ?db_first_block,
1808 ?highest_block,
1809 unwind_target = highest_block,
1810 ?segment,
1811 "Setting unwind target."
1812 );
1813 return Ok(Some(highest_block))
1814 }
1815
1816 if let Some((db_last_key, _)) = db_cursor.last()? &&
1817 highest_static_file_block
1818 .is_none_or(|highest_block| block_from_key(&db_last_key) > highest_block)
1819 {
1820 debug!(
1821 target: "reth::providers::static_file",
1822 ?segment,
1823 "Database has entries beyond static files, no unwind needed"
1824 );
1825 return Ok(None)
1826 }
1827 } else {
1828 debug!(target: "reth::providers::static_file", ?segment, "No database entries found");
1829 }
1830
1831 let highest_static_file_block = highest_static_file_block.unwrap_or_default();
1832
1833 let stage_id = segment.to_stage_id();
1834 let checkpoint_block_number =
1835 provider.get_stage_checkpoint(stage_id)?.unwrap_or_default().block_number;
1836
1837 let effective_coverage_block =
1838 Self::effective_coverage_block(provider, segment, highest_static_file_block)?;
1839
1840 if checkpoint_block_number > effective_coverage_block {
1841 info!(
1842 target: "reth::providers::static_file",
1843 checkpoint_block_number,
1844 unwind_target = effective_coverage_block,
1845 ?segment,
1846 "Setting unwind target."
1847 );
1848 return Ok(Some(effective_coverage_block))
1849 }
1850
1851 if checkpoint_block_number < highest_static_file_block {
1852 info!(
1853 target: "reth::providers",
1854 ?segment,
1855 from = highest_static_file_block,
1856 to = checkpoint_block_number,
1857 "Unwinding static file segment."
1858 );
1859 let mut writer = self.latest_writer(segment)?;
1860 match segment {
1861 StaticFileSegment::AccountChangeSets => {
1862 writer.prune_account_changesets(checkpoint_block_number)?;
1863 }
1864 StaticFileSegment::StorageChangeSets => {
1865 writer.prune_storage_changesets(checkpoint_block_number)?;
1866 }
1867 _ => unreachable!("invalid segment for changeset invariants"),
1868 }
1869 writer.commit()?;
1870 }
1871
1872 Ok(None)
1873 }
1874
1875 fn effective_coverage_block<Provider>(
1884 provider: &Provider,
1885 segment: StaticFileSegment,
1886 highest_static_file_block: BlockNumber,
1887 ) -> ProviderResult<BlockNumber>
1888 where
1889 Provider: PruneCheckpointReader,
1890 {
1891 let Some(prune_segment) = Self::prune_segment_for_static_file(segment) else {
1892 return Ok(highest_static_file_block)
1893 };
1894
1895 let prune_checkpoint_block = provider
1896 .get_prune_checkpoint(prune_segment)?
1897 .and_then(|checkpoint| checkpoint.block_number)
1898 .unwrap_or_default();
1899
1900 Ok(highest_static_file_block.max(prune_checkpoint_block))
1901 }
1902
1903 const fn prune_segment_for_static_file(segment: StaticFileSegment) -> Option<PruneSegment> {
1906 match segment {
1907 StaticFileSegment::Receipts => Some(PruneSegment::Receipts),
1908 StaticFileSegment::TransactionSenders => Some(PruneSegment::SenderRecovery),
1909 StaticFileSegment::AccountChangeSets => Some(PruneSegment::AccountHistory),
1910 StaticFileSegment::StorageChangeSets => Some(PruneSegment::StorageHistory),
1911 StaticFileSegment::Headers | StaticFileSegment::Transactions => None,
1912 }
1913 }
1914
1915 pub fn earliest_history_height(&self) -> BlockNumber {
1923 self.earliest_history_height.load(std::sync::atomic::Ordering::Relaxed)
1924 }
1925
1926 pub fn get_lowest_range(&self, segment: StaticFileSegment) -> Option<SegmentRangeInclusive> {
1930 self.indexes.read().get(segment).and_then(|index| index.min_block_range)
1931 }
1932
1933 pub fn get_lowest_range_start(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
1939 self.get_lowest_range(segment).map(|range| range.start())
1940 }
1941
1942 pub fn get_lowest_range_end(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
1948 self.get_lowest_range(segment).map(|range| range.end())
1949 }
1950
1951 pub fn get_highest_static_file_block(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
1955 self.indexes.read().get(segment).map(|index| index.max_block)
1956 }
1957
1958 fn bound_range(
1965 &self,
1966 range: impl RangeBounds<BlockNumber>,
1967 segment: StaticFileSegment,
1968 ) -> RangeInclusive<BlockNumber> {
1969 let highest_block = self.get_highest_static_file_block(segment).unwrap_or(0);
1970
1971 let start = match range.start_bound() {
1972 Bound::Included(&n) => n,
1973 Bound::Excluded(&n) => n.saturating_add(1),
1974 Bound::Unbounded => 0,
1975 };
1976 let end = match range.end_bound() {
1977 Bound::Included(&n) => n.min(highest_block),
1978 Bound::Excluded(&n) => n.saturating_sub(1).min(highest_block),
1979 Bound::Unbounded => highest_block,
1980 };
1981
1982 start..=end
1983 }
1984
1985 pub fn get_highest_static_file_tx(&self, segment: StaticFileSegment) -> Option<TxNumber> {
1989 self.indexes
1990 .read()
1991 .get(segment)
1992 .and_then(|index| index.available_block_ranges_by_max_tx.as_ref())
1993 .and_then(|index| index.last_key_value().map(|(last_tx, _)| *last_tx))
1994 }
1995
1996 pub fn get_highest_static_files(&self) -> HighestStaticFiles {
1998 HighestStaticFiles {
1999 receipts: self.get_highest_static_file_block(StaticFileSegment::Receipts),
2000 }
2001 }
2002
2003 pub fn find_static_file<T>(
2006 &self,
2007 segment: StaticFileSegment,
2008 func: impl Fn(StaticFileJarProvider<'_, N>) -> ProviderResult<Option<T>>,
2009 ) -> ProviderResult<Option<T>> {
2010 if let Some(ranges) =
2011 self.indexes.read().get(segment).map(|index| &index.expected_block_ranges_by_max_block)
2012 {
2013 for range in ranges.values().rev() {
2015 if let Some(res) = func(self.get_or_create_jar_provider(segment, range)?)? {
2016 return Ok(Some(res));
2017 }
2018 }
2019 }
2020
2021 Ok(None)
2022 }
2023
2024 pub fn fetch_range_with_predicate<T, F, P>(
2030 &self,
2031 segment: StaticFileSegment,
2032 range: Range<u64>,
2033 mut get_fn: F,
2034 mut predicate: P,
2035 ) -> ProviderResult<Vec<T>>
2036 where
2037 F: FnMut(&mut StaticFileCursor<'_>, u64) -> ProviderResult<Option<T>>,
2038 P: FnMut(&T) -> bool,
2039 {
2040 let mut result = Vec::with_capacity((range.end - range.start).min(100) as usize);
2041
2042 macro_rules! get_provider {
2046 ($number:expr) => {{
2047 match self.get_segment_provider(segment, $number) {
2048 Ok(provider) => provider,
2049 Err(
2050 ProviderError::MissingStaticFileBlock(_, _) |
2051 ProviderError::MissingStaticFileTx(_, _),
2052 ) => return Ok(result),
2053 Err(err) => return Err(err),
2054 }
2055 }};
2056 }
2057
2058 let mut provider = get_provider!(range.start);
2059 let mut cursor = provider.cursor()?;
2060
2061 'outer: for number in range {
2063 let mut retrying = false;
2067
2068 'inner: loop {
2070 match get_fn(&mut cursor, number)? {
2071 Some(res) => {
2072 if !predicate(&res) {
2073 break 'outer;
2074 }
2075 result.push(res);
2076 break 'inner;
2077 }
2078 None => {
2079 if retrying {
2080 return Ok(result);
2081 }
2082 drop(cursor);
2087 drop(provider);
2088 provider = get_provider!(number);
2089 cursor = provider.cursor()?;
2090 retrying = true;
2091 }
2092 }
2093 }
2094 }
2095
2096 result.shrink_to_fit();
2097
2098 Ok(result)
2099 }
2100
2101 pub fn fetch_range_iter<'a, T, F>(
2106 &'a self,
2107 segment: StaticFileSegment,
2108 range: Range<u64>,
2109 get_fn: F,
2110 ) -> ProviderResult<impl Iterator<Item = ProviderResult<Option<T>>> + 'a>
2111 where
2112 F: Fn(&mut StaticFileCursor<'_>, u64) -> ProviderResult<Option<T>> + 'a,
2113 T: std::fmt::Debug,
2114 {
2115 let mut provider = self.get_maybe_segment_provider(segment, range.start)?;
2116 Ok(range.map(move |number| {
2117 match provider
2118 .as_ref()
2119 .map(|provider| get_fn(&mut provider.cursor()?, number))
2120 .and_then(|result| result.transpose())
2121 {
2122 Some(result) => result.map(Some),
2123 None => {
2124 provider.take();
2128 provider = self.get_maybe_segment_provider(segment, number)?;
2129 provider
2130 .as_ref()
2131 .map(|provider| get_fn(&mut provider.cursor()?, number))
2132 .and_then(|result| result.transpose())
2133 .transpose()
2134 }
2135 }
2136 }))
2137 }
2138
2139 pub fn directory(&self) -> &Path {
2141 &self.path
2142 }
2143
2144 pub fn get_with_static_file_or_database<T, FS, FD>(
2154 &self,
2155 segment: StaticFileSegment,
2156 number: u64,
2157 fetch_from_static_file: FS,
2158 fetch_from_database: FD,
2159 ) -> ProviderResult<Option<T>>
2160 where
2161 FS: Fn(&Self) -> ProviderResult<Option<T>>,
2162 FD: Fn() -> ProviderResult<Option<T>>,
2163 {
2164 let static_file_upper_bound = if segment.is_block_or_change_based() {
2166 self.get_highest_static_file_block(segment)
2167 } else {
2168 self.get_highest_static_file_tx(segment)
2169 };
2170
2171 if static_file_upper_bound
2172 .is_some_and(|static_file_upper_bound| static_file_upper_bound >= number)
2173 {
2174 return fetch_from_static_file(self);
2175 }
2176 fetch_from_database()
2177 }
2178
2179 pub fn get_range_with_static_file_or_database<T, P, FS, FD>(
2191 &self,
2192 segment: StaticFileSegment,
2193 mut block_or_tx_range: Range<u64>,
2194 fetch_from_static_file: FS,
2195 mut fetch_from_database: FD,
2196 mut predicate: P,
2197 ) -> ProviderResult<Vec<T>>
2198 where
2199 FS: Fn(&Self, Range<u64>, &mut P) -> ProviderResult<Vec<T>>,
2200 FD: FnMut(Range<u64>, P) -> ProviderResult<Vec<T>>,
2201 P: FnMut(&T) -> bool,
2202 {
2203 let mut data = Vec::new();
2204
2205 if let Some(static_file_upper_bound) = if segment.is_block_or_change_based() {
2207 self.get_highest_static_file_block(segment)
2208 } else {
2209 self.get_highest_static_file_tx(segment)
2210 } && block_or_tx_range.start <= static_file_upper_bound
2211 {
2212 let end = block_or_tx_range.end.min(static_file_upper_bound + 1);
2213 data.extend(fetch_from_static_file(
2214 self,
2215 block_or_tx_range.start..end,
2216 &mut predicate,
2217 )?);
2218 block_or_tx_range.start = end;
2219 }
2220
2221 if block_or_tx_range.end > block_or_tx_range.start {
2222 data.extend(fetch_from_database(block_or_tx_range, predicate)?)
2223 }
2224
2225 Ok(data)
2226 }
2227
2228 #[cfg(any(test, feature = "test-utils"))]
2230 pub fn path(&self) -> &Path {
2231 &self.path
2232 }
2233
2234 #[cfg(any(test, feature = "test-utils"))]
2236 pub fn tx_index(&self, segment: StaticFileSegment) -> Option<SegmentRanges> {
2237 self.indexes
2238 .read()
2239 .get(segment)
2240 .and_then(|index| index.available_block_ranges_by_max_tx.as_ref())
2241 .cloned()
2242 }
2243
2244 #[cfg(any(test, feature = "test-utils"))]
2246 pub fn expected_block_index(&self, segment: StaticFileSegment) -> Option<SegmentRanges> {
2247 self.indexes
2248 .read()
2249 .get(segment)
2250 .map(|index| &index.expected_block_ranges_by_max_block)
2251 .cloned()
2252 }
2253}
2254
2255#[derive(Debug)]
2256struct StaticFileSegmentIndex {
2257 min_block_range: Option<SegmentRangeInclusive>,
2269 max_block: u64,
2271 expected_block_ranges_by_max_block: SegmentRanges,
2277 available_block_ranges_by_max_tx: Option<SegmentRanges>,
2284}
2285
2286pub trait StaticFileWriter {
2288 type Primitives: Send + Sync + 'static;
2290
2291 fn get_writer(
2293 &self,
2294 block: BlockNumber,
2295 segment: StaticFileSegment,
2296 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2297
2298 fn latest_writer(
2301 &self,
2302 segment: StaticFileSegment,
2303 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2304
2305 fn commit(&self) -> ProviderResult<()>;
2307
2308 fn has_unwind_queued(&self) -> bool;
2310
2311 fn finalize(&self) -> ProviderResult<()>;
2315}
2316
2317impl<N: NodePrimitives> StaticFileWriter for StaticFileProvider<N> {
2318 type Primitives = N;
2319
2320 fn get_writer(
2321 &self,
2322 block: BlockNumber,
2323 segment: StaticFileSegment,
2324 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
2325 if self.access.is_read_only() {
2326 return Err(ProviderError::ReadOnlyStaticFileAccess);
2327 }
2328
2329 trace!(target: "providers::static_file", ?block, ?segment, "Getting static file writer.");
2330 self.writers.get_or_create(segment, || {
2331 StaticFileProviderRW::new(segment, block, Arc::downgrade(&self.0), self.metrics.clone())
2332 })
2333 }
2334
2335 fn latest_writer(
2336 &self,
2337 segment: StaticFileSegment,
2338 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
2339 let genesis_number = self.0.as_ref().genesis_block_number();
2340 self.get_writer(
2341 self.get_highest_static_file_block(segment).unwrap_or(genesis_number),
2342 segment,
2343 )
2344 }
2345
2346 fn commit(&self) -> ProviderResult<()> {
2347 self.writers.commit()
2348 }
2349
2350 fn has_unwind_queued(&self) -> bool {
2351 self.writers.has_unwind_queued()
2352 }
2353
2354 fn finalize(&self) -> ProviderResult<()> {
2355 self.writers.finalize()
2356 }
2357}
2358
2359impl<N: NodePrimitives> ChangeSetReader for StaticFileProvider<N> {
2360 fn account_block_changeset(
2361 &self,
2362 block_number: BlockNumber,
2363 ) -> ProviderResult<Vec<reth_db::models::AccountBeforeTx>> {
2364 let provider = match self.get_segment_provider_for_block(
2365 StaticFileSegment::AccountChangeSets,
2366 block_number,
2367 None,
2368 ) {
2369 Ok(provider) => provider,
2370 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(Vec::new()),
2371 Err(err) => return Err(err),
2372 };
2373
2374 if let Some(offset) = provider.read_changeset_offset(block_number)? {
2375 let mut cursor = provider.cursor()?;
2376 let mut changeset = Vec::with_capacity(offset.num_changes() as usize);
2377
2378 for i in offset.changeset_range() {
2379 if let Some(change) =
2380 cursor.get_one::<reth_db::static_file::AccountChangesetMask>(i.into())?
2381 {
2382 changeset.push(change)
2383 }
2384 }
2385 Ok(changeset)
2386 } else {
2387 Ok(Vec::new())
2388 }
2389 }
2390
2391 fn get_account_before_block(
2392 &self,
2393 block_number: BlockNumber,
2394 address: Address,
2395 ) -> ProviderResult<Option<reth_db::models::AccountBeforeTx>> {
2396 let provider = match self.get_segment_provider_for_block(
2397 StaticFileSegment::AccountChangeSets,
2398 block_number,
2399 None,
2400 ) {
2401 Ok(provider) => provider,
2402 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(None),
2403 Err(err) => return Err(err),
2404 };
2405
2406 let Some(offset) = provider.read_changeset_offset(block_number)? else {
2407 return Ok(None);
2408 };
2409
2410 let mut cursor = provider.cursor()?;
2411 let range = offset.changeset_range();
2412 let mut low = range.start;
2413 let mut high = range.end;
2414
2415 while low < high {
2416 let mid = low + (high - low) / 2;
2417 if let Some(change) =
2418 cursor.get_one::<reth_db::static_file::AccountChangesetMask>(mid.into())?
2419 {
2420 if change.address < address {
2421 low = mid + 1;
2422 } else {
2423 high = mid;
2424 }
2425 } else {
2426 debug!(
2429 target: "providers::static_file",
2430 ?low,
2431 ?mid,
2432 ?high,
2433 ?range,
2434 ?block_number,
2435 ?address,
2436 "Cannot continue binary search for account changeset fetch"
2437 );
2438 low = range.end;
2439 break;
2440 }
2441 }
2442
2443 if low < range.end &&
2444 let Some(change) = cursor
2445 .get_one::<reth_db::static_file::AccountChangesetMask>(low.into())?
2446 .filter(|change| change.address == address)
2447 {
2448 return Ok(Some(change));
2449 }
2450
2451 Ok(None)
2452 }
2453
2454 fn account_changesets_range(
2455 &self,
2456 range: impl core::ops::RangeBounds<BlockNumber>,
2457 ) -> ProviderResult<Vec<(BlockNumber, reth_db::models::AccountBeforeTx)>> {
2458 let range = self.bound_range(range, StaticFileSegment::AccountChangeSets);
2459 self.walk_account_changeset_range(range).collect()
2460 }
2461}
2462
2463impl<N: NodePrimitives> StorageChangeSetReader for StaticFileProvider<N> {
2464 fn storage_changeset(
2465 &self,
2466 block_number: BlockNumber,
2467 ) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
2468 let provider = match self.get_segment_provider_for_block(
2469 StaticFileSegment::StorageChangeSets,
2470 block_number,
2471 None,
2472 ) {
2473 Ok(provider) => provider,
2474 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(Vec::new()),
2475 Err(err) => return Err(err),
2476 };
2477
2478 if let Some(offset) = provider.read_changeset_offset(block_number)? {
2479 let mut cursor = provider.cursor()?;
2480 let mut changeset = Vec::with_capacity(offset.num_changes() as usize);
2481
2482 for i in offset.changeset_range() {
2483 if let Some(change) = cursor.get_one::<StorageChangesetMask>(i.into())? {
2484 let block_address = BlockNumberAddress((block_number, change.address));
2485 let entry = StorageEntry { key: change.key, value: change.value };
2486 changeset.push((block_address, entry));
2487 }
2488 }
2489 Ok(changeset)
2490 } else {
2491 Ok(Vec::new())
2492 }
2493 }
2494
2495 fn get_storage_before_block(
2496 &self,
2497 block_number: BlockNumber,
2498 address: Address,
2499 storage_key: B256,
2500 ) -> ProviderResult<Option<StorageEntry>> {
2501 let provider = match self.get_segment_provider_for_block(
2502 StaticFileSegment::StorageChangeSets,
2503 block_number,
2504 None,
2505 ) {
2506 Ok(provider) => provider,
2507 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(None),
2508 Err(err) => return Err(err),
2509 };
2510
2511 let Some(offset) = provider.read_changeset_offset(block_number)? else {
2512 return Ok(None);
2513 };
2514
2515 let mut cursor = provider.cursor()?;
2516 let range = offset.changeset_range();
2517 let mut low = range.start;
2518 let mut high = range.end;
2519
2520 while low < high {
2521 let mid = low + (high - low) / 2;
2522 if let Some(change) = cursor.get_one::<StorageChangesetMask>(mid.into())? {
2523 match (change.address, change.key).cmp(&(address, storage_key)) {
2524 std::cmp::Ordering::Less => low = mid + 1,
2525 _ => high = mid,
2526 }
2527 } else {
2528 debug!(
2529 target: "providers::static_file",
2530 ?low,
2531 ?mid,
2532 ?high,
2533 ?range,
2534 ?block_number,
2535 ?address,
2536 ?storage_key,
2537 "Cannot continue binary search for storage changeset fetch"
2538 );
2539 low = range.end;
2540 break;
2541 }
2542 }
2543
2544 if low < range.end &&
2545 let Some(change) = cursor
2546 .get_one::<StorageChangesetMask>(low.into())?
2547 .filter(|change| change.address == address && change.key == storage_key)
2548 {
2549 return Ok(Some(StorageEntry { key: change.key, value: change.value }));
2550 }
2551
2552 Ok(None)
2553 }
2554
2555 fn storage_changesets_range(
2556 &self,
2557 range: impl RangeBounds<BlockNumber>,
2558 ) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
2559 let range = self.bound_range(range, StaticFileSegment::StorageChangeSets);
2560 self.walk_storage_changeset_range(range).collect()
2561 }
2562}
2563
2564impl<N: NodePrimitives> StaticFileProvider<N> {
2565 pub fn walk_account_changeset_range(
2575 &self,
2576 range: impl RangeBounds<BlockNumber>,
2577 ) -> StaticFileAccountChangesetWalker<Self> {
2578 StaticFileAccountChangesetWalker::new(self.clone(), range)
2579 }
2580
2581 pub fn walk_storage_changeset_range(
2583 &self,
2584 range: impl RangeBounds<BlockNumber>,
2585 ) -> StaticFileStorageChangesetWalker<Self> {
2586 StaticFileStorageChangesetWalker::new(self.clone(), range)
2587 }
2588}
2589
2590impl<N: NodePrimitives<BlockHeader: Value>> HeaderProvider for StaticFileProvider<N> {
2591 type Header = N::BlockHeader;
2592
2593 fn header(&self, block_hash: BlockHash) -> ProviderResult<Option<Self::Header>> {
2594 self.find_static_file(StaticFileSegment::Headers, |jar_provider| {
2595 Ok(jar_provider
2596 .cursor()?
2597 .get_two::<HeaderWithHashMask<Self::Header>>((&block_hash).into())?
2598 .and_then(|(header, hash)| {
2599 if hash == block_hash {
2600 return Some(header);
2601 }
2602 None
2603 }))
2604 })
2605 }
2606
2607 fn header_by_number(&self, num: BlockNumber) -> ProviderResult<Option<Self::Header>> {
2608 self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2609 .and_then(|provider| provider.header_by_number(num))
2610 .or_else(|err| {
2611 if let ProviderError::MissingStaticFileBlock(_, _) = err {
2612 Ok(None)
2613 } else {
2614 Err(err)
2615 }
2616 })
2617 }
2618
2619 fn headers_range(
2620 &self,
2621 range: impl RangeBounds<BlockNumber>,
2622 ) -> ProviderResult<Vec<Self::Header>> {
2623 self.fetch_range_with_predicate(
2624 StaticFileSegment::Headers,
2625 to_range(range),
2626 |cursor, number| cursor.get_one::<HeaderMask<Self::Header>>(number.into()),
2627 |_| true,
2628 )
2629 }
2630
2631 fn sealed_header(
2632 &self,
2633 num: BlockNumber,
2634 ) -> ProviderResult<Option<SealedHeader<Self::Header>>> {
2635 self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2636 .and_then(|provider| provider.sealed_header(num))
2637 .or_else(|err| {
2638 if let ProviderError::MissingStaticFileBlock(_, _) = err {
2639 Ok(None)
2640 } else {
2641 Err(err)
2642 }
2643 })
2644 }
2645
2646 fn sealed_headers_while(
2647 &self,
2648 range: impl RangeBounds<BlockNumber>,
2649 predicate: impl FnMut(&SealedHeader<Self::Header>) -> bool,
2650 ) -> ProviderResult<Vec<SealedHeader<Self::Header>>> {
2651 self.fetch_range_with_predicate(
2652 StaticFileSegment::Headers,
2653 to_range(range),
2654 |cursor, number| {
2655 Ok(cursor
2656 .get_two::<HeaderWithHashMask<Self::Header>>(number.into())?
2657 .map(|(header, hash)| SealedHeader::new(header, hash)))
2658 },
2659 predicate,
2660 )
2661 }
2662}
2663
2664impl<N: NodePrimitives> BlockHashReader for StaticFileProvider<N> {
2665 fn block_hash(&self, num: u64) -> ProviderResult<Option<B256>> {
2666 self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2667 .and_then(|provider| provider.block_hash(num))
2668 .or_else(|err| {
2669 if let ProviderError::MissingStaticFileBlock(_, _) = err {
2670 Ok(None)
2671 } else {
2672 Err(err)
2673 }
2674 })
2675 }
2676
2677 fn canonical_hashes_range(
2678 &self,
2679 start: BlockNumber,
2680 end: BlockNumber,
2681 ) -> ProviderResult<Vec<B256>> {
2682 self.fetch_range_with_predicate(
2683 StaticFileSegment::Headers,
2684 start..end,
2685 |cursor, number| cursor.get_one::<BlockHashMask>(number.into()),
2686 |_| true,
2687 )
2688 }
2689}
2690
2691impl<N: NodePrimitives<SignedTx: Value + SignedTransaction, Receipt: Value>> ReceiptProvider
2692 for StaticFileProvider<N>
2693{
2694 type Receipt = N::Receipt;
2695
2696 fn receipt(&self, num: TxNumber) -> ProviderResult<Option<Self::Receipt>> {
2697 self.get_segment_provider_for_transaction(StaticFileSegment::Receipts, num, None)
2698 .and_then(|provider| provider.receipt(num))
2699 .or_else(|err| {
2700 if let ProviderError::MissingStaticFileTx(_, _) = err {
2701 Ok(None)
2702 } else {
2703 Err(err)
2704 }
2705 })
2706 }
2707
2708 fn receipt_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Receipt>> {
2709 if let Some(num) = self.transaction_id(hash)? {
2710 return self.receipt(num);
2711 }
2712 Ok(None)
2713 }
2714
2715 fn receipts_by_block(
2716 &self,
2717 _block: BlockHashOrNumber,
2718 ) -> ProviderResult<Option<Vec<Self::Receipt>>> {
2719 unreachable!()
2720 }
2721
2722 fn receipts_by_tx_range(
2723 &self,
2724 range: impl RangeBounds<TxNumber>,
2725 ) -> ProviderResult<Vec<Self::Receipt>> {
2726 self.fetch_range_with_predicate(
2727 StaticFileSegment::Receipts,
2728 to_range(range),
2729 |cursor, number| cursor.get_one::<ReceiptMask<Self::Receipt>>(number.into()),
2730 |_| true,
2731 )
2732 }
2733
2734 fn receipts_by_block_range(
2735 &self,
2736 _block_range: RangeInclusive<BlockNumber>,
2737 ) -> ProviderResult<Vec<Vec<Self::Receipt>>> {
2738 Err(ProviderError::UnsupportedProvider)
2739 }
2740}
2741
2742impl<N: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>> TransactionsProviderExt
2743 for StaticFileProvider<N>
2744{
2745 fn transaction_hashes_by_range(
2746 &self,
2747 tx_range: Range<TxNumber>,
2748 ) -> ProviderResult<Vec<(TxHash, TxNumber)>> {
2749 let tx_range_size = (tx_range.end - tx_range.start) as usize;
2750
2751 let chunk_size = 100;
2755
2756 let chunks = tx_range
2758 .clone()
2759 .step_by(chunk_size)
2760 .map(|start| start..std::cmp::min(start + chunk_size as u64, tx_range.end));
2761 let mut channels = Vec::with_capacity(tx_range_size.div_ceil(chunk_size));
2762
2763 for chunk_range in chunks {
2764 let (channel_tx, channel_rx) = mpsc::channel();
2765 channels.push(channel_rx);
2766
2767 let manager = self.clone();
2768
2769 rayon::spawn(move || {
2772 let _ = manager.fetch_range_with_predicate(
2773 StaticFileSegment::Transactions,
2774 chunk_range,
2775 |cursor, number| {
2776 Ok(cursor
2777 .get_one::<TransactionMask<Self::Transaction>>(number.into())?
2778 .map(|transaction| {
2779 let _ = channel_tx.send(transaction_hash((number, transaction)));
2780 }))
2781 },
2782 |_| true,
2783 );
2784 });
2785 }
2786
2787 let mut tx_list = Vec::with_capacity(tx_range_size);
2788
2789 for channel in channels {
2791 while let Ok(tx) = channel.recv() {
2792 let (tx_hash, tx_id) = tx.map_err(|boxed| *boxed)?;
2793 tx_list.push((tx_hash, tx_id));
2794 }
2795 }
2796
2797 Ok(tx_list)
2798 }
2799}
2800
2801impl<N: NodePrimitives<SignedTx: Decompress + SignedTransaction>> TransactionsProvider
2802 for StaticFileProvider<N>
2803{
2804 type Transaction = N::SignedTx;
2805
2806 fn transaction_id(&self, tx_hash: TxHash) -> ProviderResult<Option<TxNumber>> {
2807 self.find_static_file(StaticFileSegment::Transactions, |jar_provider| {
2808 let mut cursor = jar_provider.cursor()?;
2809 if cursor
2810 .get_one::<TransactionMask<Self::Transaction>>((&tx_hash).into())?
2811 .and_then(|tx| (*tx.tx_hash() == tx_hash).then_some(tx))
2812 .is_some()
2813 {
2814 Ok(cursor.number())
2815 } else {
2816 Ok(None)
2817 }
2818 })
2819 }
2820
2821 fn transaction_by_id(&self, num: TxNumber) -> ProviderResult<Option<Self::Transaction>> {
2822 self.get_segment_provider_for_transaction(StaticFileSegment::Transactions, num, None)
2823 .and_then(|provider| provider.transaction_by_id(num))
2824 .or_else(|err| {
2825 if let ProviderError::MissingStaticFileTx(_, _) = err {
2826 Ok(None)
2827 } else {
2828 Err(err)
2829 }
2830 })
2831 }
2832
2833 fn transaction_by_id_unhashed(
2834 &self,
2835 num: TxNumber,
2836 ) -> ProviderResult<Option<Self::Transaction>> {
2837 self.get_segment_provider_for_transaction(StaticFileSegment::Transactions, num, None)
2838 .and_then(|provider| provider.transaction_by_id_unhashed(num))
2839 .or_else(|err| {
2840 if let ProviderError::MissingStaticFileTx(_, _) = err {
2841 Ok(None)
2842 } else {
2843 Err(err)
2844 }
2845 })
2846 }
2847
2848 fn transaction_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Transaction>> {
2849 self.find_static_file(StaticFileSegment::Transactions, |jar_provider| {
2850 Ok(jar_provider
2851 .cursor()?
2852 .get_one::<TransactionMask<Self::Transaction>>((&hash).into())?
2853 .and_then(|tx| (*tx.tx_hash() == hash).then_some(tx)))
2854 })
2855 }
2856
2857 fn transaction_by_hash_with_meta(
2858 &self,
2859 _hash: TxHash,
2860 ) -> ProviderResult<Option<(Self::Transaction, TransactionMeta)>> {
2861 Err(ProviderError::UnsupportedProvider)
2863 }
2864
2865 fn transactions_by_block(
2866 &self,
2867 _block_id: BlockHashOrNumber,
2868 ) -> ProviderResult<Option<Vec<Self::Transaction>>> {
2869 Err(ProviderError::UnsupportedProvider)
2871 }
2872
2873 fn transactions_by_block_range(
2874 &self,
2875 _range: impl RangeBounds<BlockNumber>,
2876 ) -> ProviderResult<Vec<Vec<Self::Transaction>>> {
2877 Err(ProviderError::UnsupportedProvider)
2879 }
2880
2881 fn transactions_by_tx_range(
2882 &self,
2883 range: impl RangeBounds<TxNumber>,
2884 ) -> ProviderResult<Vec<Self::Transaction>> {
2885 self.fetch_range_with_predicate(
2886 StaticFileSegment::Transactions,
2887 to_range(range),
2888 |cursor, number| cursor.get_one::<TransactionMask<Self::Transaction>>(number.into()),
2889 |_| true,
2890 )
2891 }
2892
2893 fn senders_by_tx_range(
2894 &self,
2895 range: impl RangeBounds<TxNumber>,
2896 ) -> ProviderResult<Vec<Address>> {
2897 self.fetch_range_with_predicate(
2898 StaticFileSegment::TransactionSenders,
2899 to_range(range),
2900 |cursor, number| cursor.get_one::<TransactionSenderMask>(number.into()),
2901 |_| true,
2902 )
2903 }
2904
2905 fn transaction_sender(&self, id: TxNumber) -> ProviderResult<Option<Address>> {
2906 self.get_segment_provider_for_transaction(StaticFileSegment::TransactionSenders, id, None)
2907 .and_then(|provider| provider.transaction_sender(id))
2908 .or_else(|err| {
2909 if let ProviderError::MissingStaticFileTx(_, _) = err {
2910 Ok(None)
2911 } else {
2912 Err(err)
2913 }
2914 })
2915 }
2916}
2917
2918impl<N: NodePrimitives> BlockNumReader for StaticFileProvider<N> {
2919 fn chain_info(&self) -> ProviderResult<ChainInfo> {
2920 Err(ProviderError::UnsupportedProvider)
2922 }
2923
2924 fn best_block_number(&self) -> ProviderResult<BlockNumber> {
2925 Err(ProviderError::UnsupportedProvider)
2927 }
2928
2929 fn last_block_number(&self) -> ProviderResult<BlockNumber> {
2930 Ok(self.get_highest_static_file_block(StaticFileSegment::Headers).unwrap_or_default())
2931 }
2932
2933 fn block_number(&self, _hash: B256) -> ProviderResult<Option<BlockNumber>> {
2934 Err(ProviderError::UnsupportedProvider)
2936 }
2937}
2938
2939impl<N: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>> BlockReader
2942 for StaticFileProvider<N>
2943{
2944 type Block = N::Block;
2945
2946 fn find_block_by_hash(
2947 &self,
2948 _hash: B256,
2949 _source: BlockSource,
2950 ) -> ProviderResult<Option<Self::Block>> {
2951 Err(ProviderError::UnsupportedProvider)
2953 }
2954
2955 fn block(&self, _id: BlockHashOrNumber) -> ProviderResult<Option<Self::Block>> {
2956 Err(ProviderError::UnsupportedProvider)
2958 }
2959
2960 fn pending_block(&self) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
2961 Err(ProviderError::UnsupportedProvider)
2963 }
2964
2965 fn pending_block_and_receipts(
2966 &self,
2967 ) -> ProviderResult<Option<(RecoveredBlock<Self::Block>, Vec<Self::Receipt>)>> {
2968 Err(ProviderError::UnsupportedProvider)
2970 }
2971
2972 fn recovered_block(
2973 &self,
2974 _id: BlockHashOrNumber,
2975 _transaction_kind: TransactionVariant,
2976 ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
2977 Err(ProviderError::UnsupportedProvider)
2979 }
2980
2981 fn sealed_block_with_senders(
2982 &self,
2983 _id: BlockHashOrNumber,
2984 _transaction_kind: TransactionVariant,
2985 ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
2986 Err(ProviderError::UnsupportedProvider)
2988 }
2989
2990 fn block_range(&self, _range: RangeInclusive<BlockNumber>) -> ProviderResult<Vec<Self::Block>> {
2991 Err(ProviderError::UnsupportedProvider)
2993 }
2994
2995 fn block_with_senders_range(
2996 &self,
2997 _range: RangeInclusive<BlockNumber>,
2998 ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
2999 Err(ProviderError::UnsupportedProvider)
3000 }
3001
3002 fn recovered_block_range(
3003 &self,
3004 _range: RangeInclusive<BlockNumber>,
3005 ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
3006 Err(ProviderError::UnsupportedProvider)
3007 }
3008
3009 fn block_by_transaction_id(&self, _id: TxNumber) -> ProviderResult<Option<BlockNumber>> {
3010 Err(ProviderError::UnsupportedProvider)
3011 }
3012}
3013
3014impl<N: NodePrimitives> BlockBodyIndicesProvider for StaticFileProvider<N> {
3015 fn block_body_indices(&self, _num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
3016 Err(ProviderError::UnsupportedProvider)
3017 }
3018
3019 fn block_body_indices_range(
3020 &self,
3021 _range: RangeInclusive<BlockNumber>,
3022 ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
3023 Err(ProviderError::UnsupportedProvider)
3024 }
3025}
3026
3027impl<N: NodePrimitives> StatsReader for StaticFileProvider<N> {
3028 fn count_entries<T: Table>(&self) -> ProviderResult<usize> {
3029 match T::NAME {
3030 tables::CanonicalHeaders::NAME |
3031 tables::Headers::<Header>::NAME |
3032 tables::HeaderTerminalDifficulties::NAME => Ok(self
3033 .get_highest_static_file_block(StaticFileSegment::Headers)
3034 .map(|block| block + 1)
3035 .unwrap_or_default()
3036 as usize),
3037 tables::Receipts::<Receipt>::NAME => Ok(self
3038 .get_highest_static_file_tx(StaticFileSegment::Receipts)
3039 .map(|receipts| receipts + 1)
3040 .unwrap_or_default() as usize),
3041 tables::Transactions::<TransactionSigned>::NAME => Ok(self
3042 .get_highest_static_file_tx(StaticFileSegment::Transactions)
3043 .map(|txs| txs + 1)
3044 .unwrap_or_default()
3045 as usize),
3046 tables::TransactionSenders::NAME => Ok(self
3047 .get_highest_static_file_tx(StaticFileSegment::TransactionSenders)
3048 .map(|txs| txs + 1)
3049 .unwrap_or_default() as usize),
3050 _ => Err(ProviderError::UnsupportedProvider),
3051 }
3052 }
3053}
3054
3055#[inline]
3057fn transaction_hash<T>(entry: (TxNumber, T)) -> Result<(B256, TxNumber), Box<ProviderError>>
3058where
3059 T: TxHashRef,
3060{
3061 let (tx_id, tx) = entry;
3062 Ok((*tx.tx_hash(), tx_id))
3063}
3064
3065#[cfg(test)]
3066mod tests {
3067 use std::collections::BTreeMap;
3068
3069 use reth_chain_state::EthPrimitives;
3070 use reth_db::test_utils::create_test_static_files_dir;
3071 use reth_static_file_types::{SegmentRangeInclusive, StaticFileSegment};
3072
3073 use crate::{providers::StaticFileProvider, StaticFileProviderBuilder};
3074
3075 #[test]
3076 fn test_find_fixed_range_with_block_index() -> eyre::Result<()> {
3077 let (static_dir, _) = create_test_static_files_dir();
3078 let sf_rw: StaticFileProvider<EthPrimitives> =
3079 StaticFileProviderBuilder::read_write(&static_dir).with_blocks_per_file(100).build()?;
3080
3081 let segment = StaticFileSegment::Headers;
3082
3083 assert_eq!(
3085 sf_rw.find_fixed_range_with_block_index(segment, None, 0),
3086 SegmentRangeInclusive::new(0, 99)
3087 );
3088 assert_eq!(
3089 sf_rw.find_fixed_range_with_block_index(segment, None, 250),
3090 SegmentRangeInclusive::new(200, 299)
3091 );
3092
3093 assert_eq!(
3095 sf_rw.find_fixed_range_with_block_index(segment, Some(&BTreeMap::new()), 150),
3096 SegmentRangeInclusive::new(100, 199)
3097 );
3098
3099 let block_index = BTreeMap::from_iter([
3101 (99, SegmentRangeInclusive::new(0, 99)),
3102 (199, SegmentRangeInclusive::new(100, 199)),
3103 (299, SegmentRangeInclusive::new(200, 299)),
3104 ]);
3105
3106 assert_eq!(
3108 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 0),
3109 SegmentRangeInclusive::new(0, 99)
3110 );
3111 assert_eq!(
3112 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 50),
3113 SegmentRangeInclusive::new(0, 99)
3114 );
3115 assert_eq!(
3116 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 99),
3117 SegmentRangeInclusive::new(0, 99)
3118 );
3119 assert_eq!(
3120 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 100),
3121 SegmentRangeInclusive::new(100, 199)
3122 );
3123 assert_eq!(
3124 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 150),
3125 SegmentRangeInclusive::new(100, 199)
3126 );
3127 assert_eq!(
3128 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 199),
3129 SegmentRangeInclusive::new(100, 199)
3130 );
3131
3132 assert_eq!(
3135 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 300),
3136 SegmentRangeInclusive::new(300, 399)
3137 );
3138 assert_eq!(
3139 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 350),
3140 SegmentRangeInclusive::new(300, 399)
3141 );
3142
3143 assert_eq!(
3145 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 500),
3146 SegmentRangeInclusive::new(500, 599)
3147 );
3148
3149 assert_eq!(
3151 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 1000),
3152 SegmentRangeInclusive::new(1000, 1099)
3153 );
3154
3155 let mixed_size_index = BTreeMap::from_iter([
3158 (49, SegmentRangeInclusive::new(0, 49)), (149, SegmentRangeInclusive::new(50, 149)), (349, SegmentRangeInclusive::new(150, 349)), ]);
3162
3163 assert_eq!(
3165 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 25),
3166 SegmentRangeInclusive::new(0, 49)
3167 );
3168 assert_eq!(
3169 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 100),
3170 SegmentRangeInclusive::new(50, 149)
3171 );
3172 assert_eq!(
3173 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 200),
3174 SegmentRangeInclusive::new(150, 349)
3175 );
3176
3177 assert_eq!(
3180 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 350),
3181 SegmentRangeInclusive::new(350, 449)
3182 );
3183 assert_eq!(
3184 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 450),
3185 SegmentRangeInclusive::new(450, 549)
3186 );
3187 assert_eq!(
3188 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 550),
3189 SegmentRangeInclusive::new(550, 649)
3190 );
3191
3192 Ok(())
3193 }
3194}