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_execution_types::RecoveredBlockAndExecutionOutput;
37use reth_nippy_jar::{NippyJar, NippyJarChecker};
38use reth_node_types::NodePrimitives;
39use reth_primitives_traits::{
40 dashmap::DashMap, AlloyBlockHeader as _, BlockBody as _, RecoveredBlock, SealedHeader,
41 SignedTransaction, StorageEntry,
42};
43use reth_prune_types::PruneSegment;
44use reth_stages_types::PipelineTarget;
45use reth_static_file_types::{
46 find_fixed_range, HighestStaticFiles, SegmentHeader, SegmentRangeInclusive, StaticFileMap,
47 StaticFileSegment, DEFAULT_BLOCKS_PER_STATIC_FILE,
48};
49use reth_storage_api::{
50 BlockBodyIndicesProvider, ChangeSetReader, DBProvider, PruneCheckpointReader,
51 StorageChangeSetReader, StorageSettingsCache,
52};
53use reth_storage_errors::provider::{ProviderError, ProviderResult, StaticFileWriterError};
54use std::{
55 collections::BTreeMap,
56 fmt::Debug,
57 ops::{Bound, Deref, Range, RangeBounds, RangeInclusive},
58 path::{Path, PathBuf},
59 sync::{
60 atomic::{AtomicU64, Ordering},
61 mpsc, Arc,
62 },
63};
64use tracing::{debug, info, info_span, instrument, trace, warn};
65
66type SegmentRanges = BTreeMap<u64, SegmentRangeInclusive>;
69
70#[derive(Debug, Default, PartialEq, Eq)]
72pub enum StaticFileAccess {
73 #[default]
75 RO,
76 RW,
78}
79
80impl StaticFileAccess {
81 pub const fn is_read_only(&self) -> bool {
83 matches!(self, Self::RO)
84 }
85
86 pub const fn is_read_write(&self) -> bool {
88 matches!(self, Self::RW)
89 }
90}
91
92#[derive(Debug, Clone, Copy, Default)]
96pub struct StaticFileWriteCtx {
97 pub write_senders: bool,
99 pub write_receipts: bool,
101 pub write_account_changesets: bool,
103 pub write_storage_changesets: bool,
105 pub tip: BlockNumber,
107 pub receipts_prune_mode: Option<reth_prune_types::PruneMode>,
109 pub receipts_prunable: bool,
111}
112
113#[derive(Debug)]
122pub struct StaticFileProvider<N>(pub(crate) Arc<StaticFileProviderInner<N>>);
123
124impl<N> Clone for StaticFileProvider<N> {
125 fn clone(&self) -> Self {
126 Self(self.0.clone())
127 }
128}
129
130#[derive(Debug)]
132pub struct StaticFileProviderBuilder<P> {
133 access: StaticFileAccess,
134 use_metrics: bool,
135 blocks_per_file: StaticFileMap<u64>,
136 path: P,
137 genesis_block_number: u64,
138}
139
140impl<P: AsRef<Path>> StaticFileProviderBuilder<P> {
141 pub fn read_write(path: P) -> Self {
143 Self {
144 path,
145 access: StaticFileAccess::RW,
146 blocks_per_file: Default::default(),
147 use_metrics: false,
148 genesis_block_number: 0,
149 }
150 }
151
152 pub fn read_only(path: P) -> Self {
154 Self {
155 path,
156 access: StaticFileAccess::RO,
157 blocks_per_file: Default::default(),
158 use_metrics: false,
159 genesis_block_number: 0,
160 }
161 }
162
163 pub fn with_blocks_per_file_for_segments(
175 mut self,
176 segments: &<StaticFileMap<u64> as Deref>::Target,
177 ) -> Self {
178 for (segment, &blocks_per_file) in segments {
179 self.blocks_per_file.insert(segment, blocks_per_file);
180 }
181 self
182 }
183
184 pub fn with_blocks_per_file(mut self, blocks_per_file: u64) -> Self {
186 for segment in StaticFileSegment::iter() {
187 self.blocks_per_file.insert(segment, blocks_per_file);
188 }
189 self
190 }
191
192 pub fn with_blocks_per_file_for_segment(
194 mut self,
195 segment: StaticFileSegment,
196 blocks_per_file: u64,
197 ) -> Self {
198 self.blocks_per_file.insert(segment, blocks_per_file);
199 self
200 }
201
202 pub const fn with_metrics(mut self) -> Self {
204 self.use_metrics = true;
205 self
206 }
207
208 pub const fn with_genesis_block_number(mut self, genesis_block_number: u64) -> Self {
221 self.genesis_block_number = genesis_block_number;
222 self
223 }
224
225 pub fn build<N: NodePrimitives>(self) -> ProviderResult<StaticFileProvider<N>> {
227 let mut provider = StaticFileProviderInner::new(self.path, self.access)?;
228 if self.use_metrics {
229 provider.metrics = Some(Arc::new(StaticFileProviderMetrics::default()));
230 }
231
232 for (segment, blocks_per_file) in *self.blocks_per_file {
233 provider.blocks_per_file.insert(segment, blocks_per_file);
234 }
235 provider.genesis_block_number = self.genesis_block_number;
236
237 let provider = StaticFileProvider(Arc::new(provider));
238 provider.initialize_index()?;
239 Ok(provider)
240 }
241}
242
243impl<N: NodePrimitives> StaticFileProvider<N> {
244 fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
246 let provider = Self(Arc::new(StaticFileProviderInner::new(path, access)?));
247 provider.initialize_index()?;
248 Ok(provider)
249 }
250}
251
252impl<N: NodePrimitives> StaticFileProvider<N> {
253 pub fn read_only(path: impl AsRef<Path>) -> ProviderResult<Self> {
258 Self::new(path, StaticFileAccess::RO)
259 }
260
261 pub fn read_write(path: impl AsRef<Path>) -> ProviderResult<Self> {
263 Self::new(path, StaticFileAccess::RW)
264 }
265}
266
267impl<N: NodePrimitives> Deref for StaticFileProvider<N> {
268 type Target = StaticFileProviderInner<N>;
269
270 fn deref(&self) -> &Self::Target {
271 &self.0
272 }
273}
274
275#[cfg(test)]
276type BeforeCacheInsert = Box<dyn FnOnce(StaticFileSegment) + Send + 'static>;
277
278#[cfg(test)]
279#[derive(Default)]
280struct BeforeCacheInsertHook(parking_lot::Mutex<Option<BeforeCacheInsert>>);
281
282#[cfg(test)]
283impl Debug for BeforeCacheInsertHook {
284 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
285 f.debug_struct("BeforeCacheInsertHook").finish_non_exhaustive()
286 }
287}
288
289#[cfg(test)]
290impl BeforeCacheInsertHook {
291 fn set(&self, hook: impl FnOnce(StaticFileSegment) + Send + 'static) {
292 let previous = self.0.lock().replace(Box::new(hook));
293 assert!(previous.is_none(), "cache fill hook already installed");
294 }
295
296 fn fire(&self, segment: StaticFileSegment) {
297 if let Some(hook) = self.0.lock().take() {
298 hook(segment);
299 }
300 }
301}
302
303#[derive(Debug)]
305pub struct StaticFileProviderInner<N> {
306 map: DashMap<(BlockNumber, StaticFileSegment), LoadedJar>,
309 cache_generation: AtomicU64,
312 indexes: RwLock<StaticFileMap<StaticFileSegmentIndex>>,
314 earliest_history_height: AtomicU64,
325 path: PathBuf,
327 writers: StaticFileWriters<N>,
329 metrics: Option<Arc<StaticFileProviderMetrics>>,
331 access: StaticFileAccess,
333 blocks_per_file: StaticFileMap<u64>,
335 _lock_file: Option<StorageLock>,
337 genesis_block_number: u64,
339 #[cfg(test)]
340 before_cache_insert: BeforeCacheInsertHook,
341 #[cfg(test)]
342 after_cache_validation: BeforeCacheInsertHook,
343}
344
345impl<N: NodePrimitives> StaticFileProviderInner<N> {
346 fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
348 let _lock_file = if access.is_read_write() {
349 StorageLock::try_acquire(path.as_ref()).map_err(ProviderError::other)?.into()
350 } else {
351 None
352 };
353
354 let mut blocks_per_file = StaticFileMap::default();
355 for segment in StaticFileSegment::iter() {
356 blocks_per_file.insert(segment, DEFAULT_BLOCKS_PER_STATIC_FILE);
357 }
358
359 let provider = Self {
360 map: Default::default(),
361 cache_generation: Default::default(),
362 indexes: Default::default(),
363 writers: Default::default(),
364 earliest_history_height: Default::default(),
365 path: path.as_ref().to_path_buf(),
366 metrics: None,
367 access,
368 blocks_per_file,
369 _lock_file,
370 genesis_block_number: 0,
371 #[cfg(test)]
372 before_cache_insert: Default::default(),
373 #[cfg(test)]
374 after_cache_validation: Default::default(),
375 };
376
377 Ok(provider)
378 }
379
380 pub const fn is_read_only(&self) -> bool {
381 self.access.is_read_only()
382 }
383
384 pub fn find_fixed_range_with_block_index(
393 &self,
394 segment: StaticFileSegment,
395 block_index: Option<&SegmentRanges>,
396 block: BlockNumber,
397 ) -> SegmentRangeInclusive {
398 let blocks_per_file =
399 self.blocks_per_file.get(segment).copied().unwrap_or(DEFAULT_BLOCKS_PER_STATIC_FILE);
400
401 if let Some(block_index) = block_index {
402 if let Some((_, range)) = block_index.range(block..).next() {
404 return *range;
406 } else if let Some((_, range)) = block_index.last_key_value() {
407 let blocks_after_last_range = block - range.end();
413 let segments_to_skip = (blocks_after_last_range - 1) / blocks_per_file;
414 let start = range.end() + 1 + segments_to_skip * blocks_per_file;
415 return SegmentRangeInclusive::new(start, start + blocks_per_file - 1);
416 }
417 }
418 find_fixed_range(block, blocks_per_file)
421 }
422
423 pub fn find_fixed_range(
436 &self,
437 segment: StaticFileSegment,
438 block: BlockNumber,
439 ) -> SegmentRangeInclusive {
440 self.find_fixed_range_with_block_index(
441 segment,
442 self.indexes.read().get(segment).map(|index| &index.expected_block_ranges_by_max_block),
443 block,
444 )
445 }
446
447 pub const fn genesis_block_number(&self) -> u64 {
449 self.genesis_block_number
450 }
451}
452
453impl<N: NodePrimitives> StaticFileProvider<N> {
454 pub fn report_metrics(&self) -> ProviderResult<()> {
459 let Some(metrics) = &self.metrics else { return Ok(()) };
460
461 let static_files = iter_static_files(&self.path).map_err(ProviderError::other)?;
462 for (segment, headers) in &*static_files {
463 let mut entries = 0;
464 let mut size = 0;
465
466 for (block_range, _) in headers {
467 let fixed_block_range = self.find_fixed_range(segment, block_range.start());
468 let jar_provider = self
469 .get_segment_provider_for_range(segment, || Some(fixed_block_range), None)?
470 .ok_or_else(|| {
471 ProviderError::MissingStaticFileBlock(segment, block_range.start())
472 })?;
473
474 entries += jar_provider.rows();
475 size += jar_provider.size() as u64;
476 }
477
478 metrics.record_segment(segment, size, headers.len(), entries);
479 }
480
481 Ok(())
482 }
483
484 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
486 fn write_headers(
487 w: &mut StaticFileProviderRWRefMut<'_, N>,
488 blocks: &[ExecutedBlock<N>],
489 ) -> ProviderResult<()> {
490 for block in blocks {
491 let b = block.recovered_block();
492 w.append_header(b.header(), &b.hash())?;
493 }
494 Ok(())
495 }
496
497 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
499 fn write_transactions(
500 w: &mut StaticFileProviderRWRefMut<'_, N>,
501 blocks: &[ExecutedBlock<N>],
502 tx_nums: &[TxNumber],
503 ) -> ProviderResult<()> {
504 for (block, &first_tx) in blocks.iter().zip(tx_nums) {
505 let b = block.recovered_block();
506 w.increment_block(b.number())?;
507 for (i, tx) in b.body().transactions().iter().enumerate() {
508 w.append_transaction(first_tx + i as u64, tx)?;
509 }
510 }
511 Ok(())
512 }
513
514 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
516 fn write_transaction_senders(
517 w: &mut StaticFileProviderRWRefMut<'_, N>,
518 blocks: &[ExecutedBlock<N>],
519 tx_nums: &[TxNumber],
520 ) -> ProviderResult<()> {
521 for (block, &first_tx) in blocks.iter().zip(tx_nums) {
522 let b = block.recovered_block();
523 w.increment_block(b.number())?;
524 for (i, sender) in b.senders_iter().enumerate() {
525 w.append_transaction_sender(first_tx + i as u64, sender)?;
526 }
527 }
528 Ok(())
529 }
530
531 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
533 fn write_receipts(
534 w: &mut StaticFileProviderRWRefMut<'_, N>,
535 blocks: &[ExecutedBlock<N>],
536 tx_nums: &[TxNumber],
537 ctx: &StaticFileWriteCtx,
538 ) -> ProviderResult<()> {
539 for (block, &first_tx) in blocks.iter().zip(tx_nums) {
540 let block_number = block.recovered_block().number();
541 w.increment_block(block_number)?;
542
543 if ctx.receipts_prunable &&
545 ctx.receipts_prune_mode
546 .is_some_and(|mode| mode.should_prune(block_number, ctx.tip))
547 {
548 continue
549 }
550
551 for (i, receipt) in block.execution_outcome().receipts.iter().enumerate() {
552 w.append_receipt(first_tx + i as u64, receipt)?;
553 }
554 }
555 Ok(())
556 }
557
558 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
560 fn write_account_changesets(
561 w: &mut StaticFileProviderRWRefMut<'_, N>,
562 blocks: &[ExecutedBlock<N>],
563 ) -> ProviderResult<()> {
564 for block in blocks {
565 let block_number = block.recovered_block().number();
566 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
567
568 let changeset: Vec<_> = reverts
569 .accounts
570 .into_iter()
571 .flatten()
572 .map(|(address, info)| AccountBeforeTx { address, info: info.map(Into::into) })
573 .collect();
574 w.append_account_changeset(changeset, block_number)?;
575 }
576 Ok(())
577 }
578
579 #[instrument(level = "debug", target = "providers::db", skip_all)]
581 fn write_storage_changesets(
582 w: &mut StaticFileProviderRWRefMut<'_, N>,
583 blocks: &[ExecutedBlock<N>],
584 ) -> ProviderResult<()> {
585 for block in blocks {
586 let block_number = block.recovered_block().number();
587 let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
588
589 let changeset: Vec<_> = reverts
590 .storage
591 .into_iter()
592 .flatten()
593 .flat_map(|revert| {
594 revert.storage_revert.into_iter().map(move |(key, revert_to_slot)| {
595 StorageBeforeTx {
596 address: revert.address,
597 key: key.into(),
598 value: revert_to_slot.to_previous_value(),
599 }
600 })
601 })
602 .collect();
603 w.append_storage_changeset(changeset, block_number)?;
604 }
605 Ok(())
606 }
607
608 #[instrument(level = "debug", target = "providers::static_file", skip_all, fields(?segment))]
613 fn write_segment<F>(
614 &self,
615 segment: StaticFileSegment,
616 first_block_number: BlockNumber,
617 f: F,
618 ) -> ProviderResult<()>
619 where
620 F: FnOnce(&mut StaticFileProviderRWRefMut<'_, N>) -> ProviderResult<()>,
621 {
622 let mut w = self.get_writer(first_block_number, segment)?;
623 f(&mut w)?;
624 w.sync_all()
625 }
626
627 #[instrument(level = "debug", target = "providers::static_file", skip_all)]
632 pub fn write_blocks_data(
633 &self,
634 blocks: &[ExecutedBlock<N>],
635 tx_nums: &[TxNumber],
636 ctx: StaticFileWriteCtx,
637 runtime: &reth_tasks::Runtime,
638 ) -> ProviderResult<()> {
639 if blocks.is_empty() {
640 return Ok(());
641 }
642
643 let first_block_number = blocks[0].recovered_block().number();
644
645 let mut r_headers = None;
646 let mut r_txs = None;
647 let mut r_senders = None;
648 let mut r_receipts = None;
649 let mut r_account_changesets = None;
650 let mut r_storage_changesets = None;
651
652 let span = tracing::Span::current();
655 runtime.storage_pool().in_place_scope(|s| {
656 s.spawn(|_| {
657 let _guard = span.enter();
658 r_headers =
659 Some(self.write_segment(StaticFileSegment::Headers, first_block_number, |w| {
660 Self::write_headers(w, blocks)
661 }));
662 });
663
664 s.spawn(|_| {
665 let _guard = span.enter();
666 r_txs = Some(self.write_segment(
667 StaticFileSegment::Transactions,
668 first_block_number,
669 |w| Self::write_transactions(w, blocks, tx_nums),
670 ));
671 });
672
673 if ctx.write_senders {
674 s.spawn(|_| {
675 let _guard = span.enter();
676 r_senders = Some(self.write_segment(
677 StaticFileSegment::TransactionSenders,
678 first_block_number,
679 |w| Self::write_transaction_senders(w, blocks, tx_nums),
680 ));
681 });
682 }
683
684 if ctx.write_receipts {
685 s.spawn(|_| {
686 let _guard = span.enter();
687 r_receipts = Some(self.write_segment(
688 StaticFileSegment::Receipts,
689 first_block_number,
690 |w| Self::write_receipts(w, blocks, tx_nums, &ctx),
691 ));
692 });
693 }
694
695 if ctx.write_account_changesets {
696 s.spawn(|_| {
697 let _guard = span.enter();
698 r_account_changesets = Some(self.write_segment(
699 StaticFileSegment::AccountChangeSets,
700 first_block_number,
701 |w| Self::write_account_changesets(w, blocks),
702 ));
703 });
704 }
705
706 if ctx.write_storage_changesets {
707 s.spawn(|_| {
708 let _guard = span.enter();
709 r_storage_changesets = Some(self.write_segment(
710 StaticFileSegment::StorageChangeSets,
711 first_block_number,
712 |w| Self::write_storage_changesets(w, blocks),
713 ));
714 });
715 }
716 });
717
718 r_headers.ok_or(StaticFileWriterError::ThreadPanic("headers"))??;
719 r_txs.ok_or(StaticFileWriterError::ThreadPanic("transactions"))??;
720 if ctx.write_senders {
721 r_senders.ok_or(StaticFileWriterError::ThreadPanic("senders"))??;
722 }
723 if ctx.write_receipts {
724 r_receipts.ok_or(StaticFileWriterError::ThreadPanic("receipts"))??;
725 }
726 if ctx.write_account_changesets {
727 r_account_changesets
728 .ok_or(StaticFileWriterError::ThreadPanic("account_changesets"))??;
729 }
730 if ctx.write_storage_changesets {
731 r_storage_changesets
732 .ok_or(StaticFileWriterError::ThreadPanic("storage_changesets"))??;
733 }
734 Ok(())
735 }
736
737 pub fn get_segment_provider(
740 &self,
741 segment: StaticFileSegment,
742 number: u64,
743 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
744 if segment.is_block_or_change_based() {
745 self.get_segment_provider_for_block(segment, number, None)
746 } else {
747 self.get_segment_provider_for_transaction(segment, number, None)
748 }
749 }
750
751 pub fn get_maybe_segment_provider(
756 &self,
757 segment: StaticFileSegment,
758 number: u64,
759 ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
760 let provider = if segment.is_block_or_change_based() {
761 self.get_segment_provider_for_block(segment, number, None)
762 } else {
763 self.get_segment_provider_for_transaction(segment, number, None)
764 };
765
766 match provider {
767 Ok(provider) => Ok(Some(provider)),
768 Err(
769 ProviderError::MissingStaticFileBlock(_, _) |
770 ProviderError::MissingStaticFileTx(_, _),
771 ) => Ok(None),
772 Err(err) => Err(err),
773 }
774 }
775
776 pub fn get_segment_provider_for_block(
778 &self,
779 segment: StaticFileSegment,
780 block: BlockNumber,
781 path: Option<&Path>,
782 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
783 self.get_segment_provider_for_range(
784 segment,
785 || self.get_segment_ranges_from_block(segment, block),
786 path,
787 )?
788 .ok_or(ProviderError::MissingStaticFileBlock(segment, block))
789 }
790
791 pub fn get_segment_provider_for_transaction(
793 &self,
794 segment: StaticFileSegment,
795 tx: TxNumber,
796 path: Option<&Path>,
797 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
798 self.get_segment_provider_for_range(
799 segment,
800 || self.get_segment_ranges_from_transaction(segment, tx),
801 path,
802 )?
803 .ok_or(ProviderError::MissingStaticFileTx(segment, tx))
804 }
805
806 pub fn get_segment_provider_for_range(
810 &self,
811 segment: StaticFileSegment,
812 fn_range: impl Fn() -> Option<SegmentRangeInclusive>,
813 path: Option<&Path>,
814 ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
815 let block_range = match path {
818 Some(path) => StaticFileSegment::parse_filename(
819 &path
820 .file_name()
821 .ok_or_else(|| {
822 ProviderError::MissingStaticFileSegmentPath(segment, path.to_path_buf())
823 })?
824 .to_string_lossy(),
825 )
826 .and_then(|(parsed_segment, block_range)| {
827 if parsed_segment == segment {
828 return Some(block_range);
829 }
830 None
831 }),
832 None => fn_range(),
833 };
834
835 if let Some(block_range) = block_range {
837 return Ok(Some(self.get_or_create_jar_provider(segment, &block_range)?));
838 }
839
840 Ok(None)
841 }
842
843 pub fn get_segment_provider_for_path(
845 &self,
846 path: &Path,
847 ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
848 StaticFileSegment::parse_filename(
849 &path
850 .file_name()
851 .ok_or_else(|| ProviderError::MissingStaticFilePath(path.to_path_buf()))?
852 .to_string_lossy(),
853 )
854 .map(|(segment, block_range)| self.get_or_create_jar_provider(segment, &block_range))
855 .transpose()
856 }
857
858 pub fn remove_cached_provider(
862 &self,
863 segment: StaticFileSegment,
864 fixed_block_range_end: BlockNumber,
865 ) {
866 self.map.remove(&(fixed_block_range_end, segment));
867 }
868
869 pub fn delete_segment_below_block(
886 &self,
887 segment: StaticFileSegment,
888 block: BlockNumber,
889 ) -> ProviderResult<Vec<SegmentHeader>> {
890 if block == 0 {
892 return Ok(Vec::new());
893 }
894
895 let highest_block = self.get_highest_static_file_block(segment);
896 let mut deleted_headers = Vec::new();
897
898 loop {
899 let Some(block_height) = self.get_lowest_range_end(segment) else {
900 return Ok(deleted_headers);
901 };
902
903 if block_height >= block || Some(block_height) == highest_block {
905 return Ok(deleted_headers);
906 }
907
908 debug!(
909 target: "providers::static_file",
910 ?segment,
911 ?block_height,
912 "Deleting static file below block"
913 );
914
915 let header = self.delete_jar(segment, block_height).inspect_err(|err| {
918 warn!( target: "providers::static_file", ?segment, %block_height, ?err, "Failed to delete static file below block")
919 })?;
920
921 deleted_headers.push(header);
922 }
923 }
924
925 pub fn delete_jar(
933 &self,
934 segment: StaticFileSegment,
935 block: BlockNumber,
936 ) -> ProviderResult<SegmentHeader> {
937 let fixed_block_range = self.find_fixed_range(segment, block);
938 let key = (fixed_block_range.end(), segment);
939 let file = self.path.join(segment.filename(&fixed_block_range));
940 let jar = if let Some((_, jar)) = self.map.remove(&key) {
941 jar.jar
942 } else {
943 debug!(
944 target: "providers::static_file",
945 ?file,
946 ?fixed_block_range,
947 ?block,
948 "Loading static file jar for deletion"
949 );
950 NippyJar::<SegmentHeader>::load(&file).map_err(ProviderError::other)?
951 };
952
953 let header = jar.user_header().clone();
954
955 if segment.is_change_based() {
957 let csoff_path = file.with_extension("csoff");
958 if csoff_path.exists() {
959 std::fs::remove_file(&csoff_path).map_err(ProviderError::other)?;
960 }
961 }
962
963 jar.delete().map_err(ProviderError::other)?;
964
965 self.initialize_index()?;
968
969 Ok(header)
970 }
971
972 pub fn delete_segment(&self, segment: StaticFileSegment) -> ProviderResult<Vec<SegmentHeader>> {
980 let mut deleted_headers = Vec::new();
981
982 self.writers.remove(segment);
983
984 while let Some(block_height) = self.get_highest_static_file_block(segment) {
985 debug!(
986 target: "providers::static_file",
987 ?segment,
988 ?block_height,
989 "Deleting static file jar"
990 );
991
992 let header = self.delete_jar(segment, block_height).inspect_err(|err| {
993 warn!(target: "providers::static_file", ?segment, %block_height, ?err, "Failed to delete static file jar")
994 })?;
995
996 deleted_headers.push(header);
997 }
998
999 Ok(deleted_headers)
1000 }
1001
1002 fn get_or_create_jar_provider(
1006 &self,
1007 segment: StaticFileSegment,
1008 fixed_block_range: &SegmentRangeInclusive,
1009 ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
1010 let key = (fixed_block_range.end(), segment);
1011
1012 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Getting provider");
1014 let mut provider: StaticFileJarProvider<'_, N> = if let Some(jar) = self.map.get(&key) {
1015 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Jar found in cache");
1016 jar.into()
1017 } else {
1018 let generation = self.cache_generation.load(Ordering::Acquire);
1019 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, generation, "Creating jar from scratch");
1020 let path = self.path.join(segment.filename(fixed_block_range));
1021 let jar = NippyJar::load(&path).map_err(ProviderError::other)?;
1022 let loaded = LoadedJar::new(jar)?;
1023 #[cfg(test)]
1024 self.before_cache_insert.fire(segment);
1025
1026 self.map
1030 .entry(key)
1031 .or_try_insert_with(|| -> ProviderResult<_> {
1032 let loaded = if self.cache_generation.load(Ordering::Acquire) == generation {
1036 loaded
1037 } else {
1038 trace!(target: "providers::static_file", ?segment, ?fixed_block_range, generation, "Reloading jar after cache invalidation");
1039 drop(loaded);
1040 LoadedJar::new(NippyJar::load(&path).map_err(ProviderError::other)?)?
1041 };
1042 #[cfg(test)]
1043 self.after_cache_validation.fire(segment);
1044 Ok(loaded)
1045 })?
1046 .downgrade()
1047 .into()
1048 };
1049
1050 if let Some(metrics) = &self.metrics {
1051 provider = provider.with_metrics(metrics.clone());
1052 }
1053 Ok(provider)
1054 }
1055
1056 fn get_segment_ranges_from_block(
1059 &self,
1060 segment: StaticFileSegment,
1061 block: u64,
1062 ) -> Option<SegmentRangeInclusive> {
1063 let indexes = self.indexes.read();
1064 let index = indexes.get(segment)?;
1065
1066 (index.max_block >= block).then(|| {
1067 self.find_fixed_range_with_block_index(
1068 segment,
1069 Some(&index.expected_block_ranges_by_max_block),
1070 block,
1071 )
1072 })
1073 }
1074
1075 fn get_segment_ranges_from_transaction(
1078 &self,
1079 segment: StaticFileSegment,
1080 tx: u64,
1081 ) -> Option<SegmentRangeInclusive> {
1082 let indexes = self.indexes.read();
1083 let index = indexes.get(segment)?;
1084 let available_block_ranges_by_max_tx = index.available_block_ranges_by_max_tx.as_ref()?;
1085
1086 let mut static_files_rev_iter = available_block_ranges_by_max_tx.iter().rev().peekable();
1089
1090 while let Some((tx_end, block_range)) = static_files_rev_iter.next() {
1091 if tx > *tx_end {
1092 return None;
1094 }
1095 let tx_start = static_files_rev_iter.peek().map(|(tx_end, _)| *tx_end + 1).unwrap_or(0);
1096 if tx_start <= tx {
1097 return Some(self.find_fixed_range_with_block_index(
1098 segment,
1099 Some(&index.expected_block_ranges_by_max_block),
1100 block_range.end(),
1101 ));
1102 }
1103 }
1104 None
1105 }
1106
1107 pub fn update_index(
1114 &self,
1115 segment: StaticFileSegment,
1116 segment_max_block: Option<BlockNumber>,
1117 ) -> ProviderResult<()> {
1118 trace!(
1119 target: "providers::static_file",
1120 ?segment,
1121 ?segment_max_block,
1122 "Updating provider index"
1123 );
1124 let mut indexes = self.indexes.write();
1125
1126 match segment_max_block {
1127 Some(segment_max_block) => {
1128 let fixed_range = self.find_fixed_range_with_block_index(
1129 segment,
1130 indexes.get(segment).map(|index| &index.expected_block_ranges_by_max_block),
1131 segment_max_block,
1132 );
1133
1134 let jar = NippyJar::<SegmentHeader>::load(
1135 &self.path.join(segment.filename(&fixed_range)),
1136 )
1137 .map_err(ProviderError::other)?;
1138
1139 let index = indexes
1140 .entry(segment)
1141 .and_modify(|index| {
1142 index.max_block = segment_max_block;
1144
1145 index
1149 .expected_block_ranges_by_max_block
1150 .retain(|_, block_range| block_range.start() < fixed_range.start());
1151 index
1153 .expected_block_ranges_by_max_block
1154 .insert(fixed_range.end(), fixed_range);
1155 })
1156 .or_insert_with(|| StaticFileSegmentIndex {
1157 min_block_range: None,
1158 max_block: segment_max_block,
1159 expected_block_ranges_by_max_block: BTreeMap::from([(
1160 fixed_range.end(),
1161 fixed_range,
1162 )]),
1163 available_block_ranges_by_max_tx: None,
1164 });
1165
1166 if let Some(current_block_range) = jar.user_header().block_range() {
1182 if let Some(min_block_range) = index.min_block_range.as_mut() {
1183 if current_block_range.start() == min_block_range.start() {
1186 *min_block_range = current_block_range;
1187 }
1188 } else {
1189 index.min_block_range = Some(current_block_range);
1190 }
1191 }
1192
1193 if let Some(tx_range) = jar.user_header().tx_range() {
1196 if let Some(current_block_range) = jar.user_header().block_range() {
1199 let tx_end = tx_range.end();
1200
1201 if let Some(index) = index.available_block_ranges_by_max_tx.as_mut() {
1210 index
1211 .retain(|_, block_range| block_range.start() < fixed_range.start());
1212 index.insert(tx_end, current_block_range);
1213 } else {
1214 index.available_block_ranges_by_max_tx =
1215 Some(BTreeMap::from([(tx_end, current_block_range)]));
1216 }
1217 }
1218 } else if segment.is_tx_based() {
1219 if let Some(index) = index.available_block_ranges_by_max_tx.as_mut() {
1223 index.retain(|_, block_range| block_range.start() < fixed_range.start());
1224 }
1225
1226 index.available_block_ranges_by_max_tx.take_if(|index| index.is_empty());
1228 }
1229
1230 trace!(target: "providers::static_file", ?segment, "Inserting updated jar into cache");
1232 self.map.insert((fixed_range.end(), segment), LoadedJar::new(jar)?);
1233
1234 trace!(target: "providers::static_file", ?segment, "Cleaning up jar map");
1236 self.map.retain(|(end, seg), _| !(*seg == segment && *end > fixed_range.end()));
1237 }
1238 None => {
1239 debug!(target: "providers::static_file", ?segment, "Removing segment from index");
1240 indexes.remove(segment);
1241 }
1242 };
1243
1244 trace!(target: "providers::static_file", ?segment, "Updated provider index");
1245 Ok(())
1246 }
1247
1248 pub fn initialize_index(&self) -> ProviderResult<()> {
1250 let mut indexes = self.indexes.write();
1251 indexes.clear();
1252
1253 for (segment, headers) in &*iter_static_files(&self.path).map_err(ProviderError::other)? {
1254 let min_block_range = Some(headers.first().expect("headers are not empty").0);
1259 let max_block = headers.last().expect("headers are not empty").0.end();
1260
1261 let mut expected_block_ranges_by_max_block = BTreeMap::default();
1262 let mut available_block_ranges_by_max_tx = None;
1263
1264 for (block_range, header) in headers {
1265 expected_block_ranges_by_max_block
1267 .insert(header.expected_block_end(), header.expected_block_range());
1268
1269 if let Some(tx_range) = header.tx_range() {
1271 let tx_end = tx_range.end();
1272
1273 available_block_ranges_by_max_tx
1274 .get_or_insert_with(BTreeMap::default)
1275 .insert(tx_end, *block_range);
1276 }
1277 }
1278
1279 indexes.insert(
1280 segment,
1281 StaticFileSegmentIndex {
1282 min_block_range,
1283 max_block,
1284 expected_block_ranges_by_max_block,
1285 available_block_ranges_by_max_tx,
1286 },
1287 );
1288 }
1289
1290 self.cache_generation.fetch_add(1, Ordering::AcqRel);
1294 self.map.clear();
1295
1296 if let Some(lowest_range) =
1298 indexes.get(StaticFileSegment::Transactions).and_then(|index| index.min_block_range)
1299 {
1300 self.earliest_history_height
1302 .store(lowest_range.start(), std::sync::atomic::Ordering::Relaxed);
1303 }
1304
1305 Ok(())
1306 }
1307
1308 #[instrument(skip(self, provider), fields(read_only = self.is_read_only()))]
1332 pub fn check_consistency<Provider>(
1333 &self,
1334 provider: &Provider,
1335 ) -> ProviderResult<Option<PipelineTarget>>
1336 where
1337 Provider: DBProvider
1338 + BlockReader
1339 + StageCheckpointReader
1340 + PruneCheckpointReader
1341 + ChainSpecProvider
1342 + StorageSettingsCache,
1343 N: NodePrimitives<Receipt: Value, BlockHeader: Value, SignedTx: Value>,
1344 {
1345 if provider.chain_spec().is_optimism() &&
1352 reth_chainspec::Chain::optimism_mainnet() == provider.chain_spec().chain_id()
1353 {
1354 const OVM_HEADER_1_HASH: B256 =
1356 b256!("0xbee7192e575af30420cae0c7776304ac196077ee72b048970549e4f08e875453");
1357 if provider.block_number(OVM_HEADER_1_HASH)?.is_some() {
1358 info!(target: "reth::cli",
1359 "Skipping storage verification for OP mainnet, expected inconsistency in OVM chain"
1360 );
1361 return Ok(None);
1362 }
1363 }
1364
1365 info!(target: "reth::cli", "Verifying storage consistency.");
1366
1367 let mut unwind_target: Option<BlockNumber> = None;
1368
1369 let mut update_unwind_target = |new_target| {
1370 unwind_target =
1371 unwind_target.map(|current| current.min(new_target)).or(Some(new_target));
1372 };
1373
1374 for segment in self.segments_to_check(provider) {
1375 let span = info_span!(
1376 "Checking consistency for segment",
1377 ?segment,
1378 initial_highest_block = tracing::field::Empty,
1379 highest_block = tracing::field::Empty,
1380 highest_tx = tracing::field::Empty,
1381 );
1382 let _guard = span.enter();
1383
1384 debug!(target: "reth::providers::static_file", "Checking consistency for segment");
1385
1386 let (initial_highest_block, mut highest_block) = self.maybe_heal_segment(segment)?;
1388 span.record("initial_highest_block", initial_highest_block);
1389 span.record("highest_block", highest_block);
1390
1391 if initial_highest_block != highest_block {
1396 info!(
1397 target: "reth::providers::static_file",
1398 unwind_target = highest_block,
1399 "Setting unwind target."
1400 );
1401 update_unwind_target(highest_block.unwrap_or_default());
1402 }
1403
1404 let highest_tx = self.get_highest_static_file_tx(segment);
1410 span.record("highest_tx", highest_tx);
1411 debug!(target: "reth::providers::static_file", "Checking tx index segment");
1412
1413 if let Some(highest_tx) = highest_tx {
1414 let mut last_block = highest_block.unwrap_or_default();
1415 debug!(target: "reth::providers::static_file", last_block, highest_tx, "Verifying last transaction matches last block indices");
1416 loop {
1417 let Some(indices) = provider.block_body_indices(last_block)? else {
1418 debug!(target: "reth::providers::static_file", last_block, "Block body indices not found, static files ahead of database");
1419 break
1423 };
1424
1425 debug!(target: "reth::providers::static_file", last_block, last_tx_num = indices.last_tx_num(), "Found block body indices");
1426
1427 if indices.last_tx_num() <= highest_tx {
1428 break
1429 }
1430
1431 if last_block == 0 {
1432 debug!(target: "reth::providers::static_file", "Reached block 0 in verification loop");
1433 break
1434 }
1435
1436 last_block -= 1;
1437
1438 info!(
1439 target: "reth::providers::static_file",
1440 highest_block = self.get_highest_static_file_block(segment),
1441 unwind_target = last_block,
1442 "Setting unwind target."
1443 );
1444 span.record("highest_block", last_block);
1445 highest_block = Some(last_block);
1446 update_unwind_target(last_block);
1447 }
1448 }
1449
1450 debug!(target: "reth::providers::static_file", "Ensuring invariants for segment");
1451
1452 match self.ensure_invariants_for(provider, segment, highest_tx, highest_block)? {
1453 Some(unwind) => {
1454 debug!(target: "reth::providers::static_file", unwind_target=unwind, "Invariants check returned unwind target");
1455 update_unwind_target(unwind);
1456 }
1457 None => {
1458 debug!(target: "reth::providers::static_file", "Invariants check completed, no unwind needed")
1459 }
1460 }
1461 }
1462
1463 Ok(unwind_target.map(PipelineTarget::Unwind))
1464 }
1465
1466 pub fn check_file_consistency<Provider>(&self, provider: &Provider) -> ProviderResult<()>
1475 where
1476 Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1477 {
1478 info!(target: "reth::cli", "Healing static file inconsistencies.");
1479
1480 for segment in self.segments_to_check(provider) {
1481 let _guard = info_span!("healing_static_file_segment", ?segment).entered();
1482 let _ = self.maybe_heal_segment(segment)?;
1483 }
1484
1485 Ok(())
1486 }
1487
1488 fn segments_to_check<'a, Provider>(
1490 &'a self,
1491 provider: &'a Provider,
1492 ) -> impl Iterator<Item = StaticFileSegment> + 'a
1493 where
1494 Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1495 {
1496 StaticFileSegment::iter()
1497 .filter(move |segment| self.should_check_segment(provider, *segment))
1498 }
1499
1500 fn should_check_segment<Provider>(
1502 &self,
1503 provider: &Provider,
1504 segment: StaticFileSegment,
1505 ) -> bool
1506 where
1507 Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1508 {
1509 match segment {
1510 StaticFileSegment::Headers | StaticFileSegment::Transactions => true,
1511 StaticFileSegment::Receipts => {
1512 if EitherWriter::receipts_destination(provider).is_database() {
1513 debug!(target: "reth::providers::static_file", ?segment, "Skipping receipts segment: receipts stored in database");
1516 return false;
1517 }
1518
1519 if NamedChain::Gnosis == provider.chain_spec().chain_id() ||
1520 NamedChain::Chiado == provider.chain_spec().chain_id()
1521 {
1522 debug!(target: "reth::providers::static_file", ?segment, "Skipping receipts segment: broken historical import for gnosis/chiado");
1526 return false;
1527 }
1528
1529 true
1530 }
1531 StaticFileSegment::TransactionSenders => {
1532 if EitherWriterDestination::senders(provider).is_database() {
1533 debug!(target: "reth::providers::static_file", ?segment, "Skipping senders segment: senders stored in database");
1534 return false;
1535 }
1536
1537 if Self::is_segment_fully_pruned(provider, PruneSegment::SenderRecovery) {
1538 debug!(target: "reth::providers::static_file", ?segment, "Skipping senders segment: fully pruned");
1539 return false;
1540 }
1541
1542 true
1543 }
1544 StaticFileSegment::AccountChangeSets => {
1545 if EitherWriter::account_changesets_destination(provider).is_database() {
1546 debug!(target: "reth::providers::static_file", ?segment, "Skipping account changesets segment: changesets stored in database");
1547 return false;
1548 }
1549 true
1550 }
1551 StaticFileSegment::StorageChangeSets => {
1552 if EitherWriter::storage_changesets_destination(provider).is_database() {
1553 debug!(target: "reth::providers::static_file", ?segment, "Skipping storage changesets segment: changesets stored in database");
1554 return false
1555 }
1556 true
1557 }
1558 }
1559 }
1560
1561 fn is_segment_fully_pruned<Provider>(provider: &Provider, segment: PruneSegment) -> bool
1565 where
1566 Provider: PruneCheckpointReader,
1567 {
1568 provider
1569 .get_prune_checkpoint(segment)
1570 .ok()
1571 .flatten()
1572 .is_some_and(|checkpoint| checkpoint.prune_mode.is_full())
1573 }
1574
1575 fn next_tx_num_after<Provider>(
1580 provider: &Provider,
1581 block: BlockNumber,
1582 ) -> ProviderResult<Option<TxNumber>>
1583 where
1584 Provider: BlockReader + PruneCheckpointReader,
1585 {
1586 if let Some(indices) = provider.block_body_indices(block)? {
1587 return Ok(Some(indices.next_tx_num()))
1588 }
1589 Ok(provider
1590 .get_prune_checkpoint(PruneSegment::Bodies)?
1591 .filter(|pruned| {
1592 pruned.tx_number.is_none() &&
1593 pruned.block_number.is_some_and(|pruned| pruned >= block)
1594 })
1595 .map(|_| 0))
1596 }
1597
1598 fn check_segment_consistency(&self, segment: StaticFileSegment) -> ProviderResult<()> {
1603 debug!(target: "reth::providers::static_file", "Checking segment consistency");
1604 if let Some(latest_block) = self.get_highest_static_file_block(segment) {
1605 let file_path = self
1606 .directory()
1607 .join(segment.filename(&self.find_fixed_range(segment, latest_block)));
1608 debug!(target: "reth::providers::static_file", ?file_path, latest_block, "Loading NippyJar for consistency check");
1609
1610 let jar = NippyJar::<SegmentHeader>::load(&file_path).map_err(ProviderError::other)?;
1611 debug!(target: "reth::providers::static_file", "NippyJar loaded, checking consistency");
1612
1613 NippyJarChecker::new(jar).check_consistency().map_err(ProviderError::other)?;
1614 debug!(target: "reth::providers::static_file", "NippyJar consistency check passed");
1615 } else {
1616 debug!(target: "reth::providers::static_file", "No static file block found, skipping consistency check");
1617 }
1618 Ok(())
1619 }
1620
1621 fn maybe_heal_segment(
1637 &self,
1638 segment: StaticFileSegment,
1639 ) -> ProviderResult<(Option<BlockNumber>, Option<BlockNumber>)> {
1640 let initial_highest_block = self.get_highest_static_file_block(segment);
1641 debug!(target: "reth::providers::static_file", ?initial_highest_block, "Initial highest block for segment");
1642
1643 if self.access.is_read_only() {
1644 debug!(target: "reth::providers::static_file", "Checking segment consistency (read-only)");
1647 self.check_segment_consistency(segment)?;
1648 } else {
1649 debug!(target: "reth::providers::static_file", "Fetching latest writer which might heal any potential inconsistency");
1652 self.latest_writer(segment)?;
1653 }
1654
1655 let highest_block = self.get_highest_static_file_block(segment);
1658
1659 Ok((initial_highest_block, highest_block))
1660 }
1661
1662 fn ensure_invariants_for<Provider>(
1664 &self,
1665 provider: &Provider,
1666 segment: StaticFileSegment,
1667 highest_tx: Option<u64>,
1668 highest_block: Option<BlockNumber>,
1669 ) -> ProviderResult<Option<BlockNumber>>
1670 where
1671 Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1672 N: NodePrimitives<Receipt: Value, BlockHeader: Value, SignedTx: Value>,
1673 {
1674 match segment {
1675 StaticFileSegment::Headers => self
1676 .ensure_invariants::<_, tables::Headers<N::BlockHeader>>(
1677 provider,
1678 segment,
1679 highest_block,
1680 highest_block,
1681 ),
1682 StaticFileSegment::Transactions => self
1683 .ensure_invariants::<_, tables::Transactions<N::SignedTx>>(
1684 provider,
1685 segment,
1686 highest_tx,
1687 highest_block,
1688 ),
1689 StaticFileSegment::Receipts => self
1690 .ensure_invariants::<_, tables::Receipts<N::Receipt>>(
1691 provider,
1692 segment,
1693 highest_tx,
1694 highest_block,
1695 ),
1696 StaticFileSegment::TransactionSenders => self
1697 .ensure_invariants::<_, tables::TransactionSenders>(
1698 provider,
1699 segment,
1700 highest_tx,
1701 highest_block,
1702 ),
1703 StaticFileSegment::AccountChangeSets => self
1704 .ensure_invariants::<_, tables::AccountChangeSets>(
1705 provider,
1706 segment,
1707 highest_tx,
1708 highest_block,
1709 ),
1710 StaticFileSegment::StorageChangeSets => self
1711 .ensure_changeset_invariants_by_block::<_, tables::StorageChangeSets, _>(
1712 provider,
1713 segment,
1714 highest_block,
1715 |key| key.block_number(),
1716 ),
1717 }
1718 }
1719
1720 #[instrument(skip(self, provider, segment), fields(table = T::NAME))]
1735 fn ensure_invariants<Provider, T: Table<Key = u64>>(
1736 &self,
1737 provider: &Provider,
1738 segment: StaticFileSegment,
1739 highest_static_file_entry: Option<u64>,
1740 highest_static_file_block: Option<BlockNumber>,
1741 ) -> ProviderResult<Option<BlockNumber>>
1742 where
1743 Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1744 {
1745 debug!(target: "reth::providers::static_file", "Ensuring invariants");
1746 let mut db_cursor = provider.tx_ref().cursor_read::<T>()?;
1747
1748 if let Some((db_first_entry, _)) = db_cursor.first()? {
1749 debug!(target: "reth::providers::static_file", db_first_entry, "Found first database entry");
1750 if let (Some(highest_entry), Some(highest_block)) =
1751 (highest_static_file_entry, highest_static_file_block)
1752 {
1753 if !(db_first_entry <= highest_entry || highest_entry + 1 == db_first_entry) {
1757 info!(
1758 target: "reth::providers::static_file",
1759 ?db_first_entry,
1760 ?highest_entry,
1761 unwind_target = highest_block,
1762 "Setting unwind target."
1763 );
1764 return Ok(Some(highest_block));
1765 }
1766 }
1767
1768 if let Some((db_last_entry, _)) = db_cursor.last()? &&
1769 highest_static_file_entry
1770 .is_none_or(|highest_entry| db_last_entry > highest_entry)
1771 {
1772 debug!(target: "reth::providers::static_file", db_last_entry, "Database has entries beyond static files, no unwind needed");
1773 return Ok(None)
1774 }
1775 } else {
1776 debug!(target: "reth::providers::static_file", "No database entries found");
1777 }
1778
1779 let highest_static_file_entry = highest_static_file_entry.unwrap_or_default();
1780 let highest_static_file_block = highest_static_file_block.unwrap_or_default();
1781
1782 let stage_id = segment.to_stage_id();
1785 let checkpoint_block_number =
1786 provider.get_stage_checkpoint(stage_id)?.unwrap_or_default().block_number;
1787 debug!(target: "reth::providers::static_file", ?stage_id, checkpoint_block_number, "Retrieved stage checkpoint");
1788
1789 let effective_coverage_block =
1790 Self::effective_coverage_block(provider, segment, highest_static_file_block)?;
1791
1792 if checkpoint_block_number > effective_coverage_block {
1794 info!(
1795 target: "reth::providers::static_file",
1796 checkpoint_block_number,
1797 unwind_target = effective_coverage_block,
1798 "Setting unwind target."
1799 );
1800 return Ok(Some(effective_coverage_block));
1801 }
1802
1803 if checkpoint_block_number >= highest_static_file_block {
1805 debug!(target: "reth::providers::static_file", "Invariants ensured, returning None");
1806 return Ok(None);
1807 }
1808
1809 info!(
1815 target: "reth::providers",
1816 from = highest_static_file_block,
1817 to = checkpoint_block_number,
1818 "Unwinding static file segment."
1819 );
1820 let mut writer = self.latest_writer(segment)?;
1821
1822 match segment {
1823 StaticFileSegment::Headers => {
1824 let prune_count = highest_static_file_block - checkpoint_block_number;
1825 debug!(target: "reth::providers::static_file", prune_count, "Pruning headers");
1826 writer.prune_headers(prune_count)?;
1828 }
1829 StaticFileSegment::Transactions |
1830 StaticFileSegment::Receipts |
1831 StaticFileSegment::TransactionSenders => {
1832 if let Some(next_tx_num) =
1833 Self::next_tx_num_after(provider, checkpoint_block_number)?
1834 {
1835 let number =
1838 highest_static_file_entry.saturating_add(1).saturating_sub(next_tx_num);
1839 debug!(target: "reth::providers::static_file", prune_count = number, checkpoint_block_number, "Pruning transaction based segment");
1840
1841 match segment {
1842 StaticFileSegment::Transactions => {
1843 writer.prune_transactions(number, checkpoint_block_number)?
1844 }
1845 StaticFileSegment::Receipts => {
1846 writer.prune_receipts(number, checkpoint_block_number)?
1847 }
1848 StaticFileSegment::TransactionSenders => {
1849 writer.prune_transaction_senders(number, checkpoint_block_number)?
1850 }
1851 StaticFileSegment::Headers |
1852 StaticFileSegment::AccountChangeSets |
1853 StaticFileSegment::StorageChangeSets => {
1854 unreachable!()
1855 }
1856 }
1857 } else {
1858 debug!(target: "reth::providers::static_file", checkpoint_block_number, "No block body indices found for checkpoint block");
1859 }
1860 }
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 }
1868
1869 debug!(target: "reth::providers::static_file", "Committing writer after pruning");
1870 writer.commit()?;
1871 debug!(target: "reth::providers::static_file", "Writer committed successfully");
1872
1873 debug!(target: "reth::providers::static_file", "Invariants ensured, returning None");
1874 Ok(None)
1875 }
1876
1877 fn ensure_changeset_invariants_by_block<Provider, T, F>(
1878 &self,
1879 provider: &Provider,
1880 segment: StaticFileSegment,
1881 highest_static_file_block: Option<BlockNumber>,
1882 block_from_key: F,
1883 ) -> ProviderResult<Option<BlockNumber>>
1884 where
1885 Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1886 T: Table,
1887 F: Fn(&T::Key) -> BlockNumber,
1888 {
1889 debug!(
1890 target: "reth::providers::static_file",
1891 ?segment,
1892 ?highest_static_file_block,
1893 "Ensuring changeset invariants"
1894 );
1895 let mut db_cursor = provider.tx_ref().cursor_read::<T>()?;
1896
1897 if let Some((db_first_key, _)) = db_cursor.first()? {
1898 let db_first_block = block_from_key(&db_first_key);
1899 if let Some(highest_block) = highest_static_file_block &&
1900 !(db_first_block <= highest_block || highest_block + 1 == db_first_block)
1901 {
1902 info!(
1903 target: "reth::providers::static_file",
1904 ?db_first_block,
1905 ?highest_block,
1906 unwind_target = highest_block,
1907 ?segment,
1908 "Setting unwind target."
1909 );
1910 return Ok(Some(highest_block))
1911 }
1912
1913 if let Some((db_last_key, _)) = db_cursor.last()? &&
1914 highest_static_file_block
1915 .is_none_or(|highest_block| block_from_key(&db_last_key) > highest_block)
1916 {
1917 debug!(
1918 target: "reth::providers::static_file",
1919 ?segment,
1920 "Database has entries beyond static files, no unwind needed"
1921 );
1922 return Ok(None)
1923 }
1924 } else {
1925 debug!(target: "reth::providers::static_file", ?segment, "No database entries found");
1926 }
1927
1928 let highest_static_file_block = highest_static_file_block.unwrap_or_default();
1929
1930 let stage_id = segment.to_stage_id();
1931 let checkpoint_block_number =
1932 provider.get_stage_checkpoint(stage_id)?.unwrap_or_default().block_number;
1933
1934 let effective_coverage_block =
1935 Self::effective_coverage_block(provider, segment, highest_static_file_block)?;
1936
1937 if checkpoint_block_number > effective_coverage_block {
1938 info!(
1939 target: "reth::providers::static_file",
1940 checkpoint_block_number,
1941 unwind_target = effective_coverage_block,
1942 ?segment,
1943 "Setting unwind target."
1944 );
1945 return Ok(Some(effective_coverage_block))
1946 }
1947
1948 if checkpoint_block_number < highest_static_file_block {
1949 info!(
1950 target: "reth::providers",
1951 ?segment,
1952 from = highest_static_file_block,
1953 to = checkpoint_block_number,
1954 "Unwinding static file segment."
1955 );
1956 let mut writer = self.latest_writer(segment)?;
1957 match segment {
1958 StaticFileSegment::AccountChangeSets => {
1959 writer.prune_account_changesets(checkpoint_block_number)?;
1960 }
1961 StaticFileSegment::StorageChangeSets => {
1962 writer.prune_storage_changesets(checkpoint_block_number)?;
1963 }
1964 _ => unreachable!("invalid segment for changeset invariants"),
1965 }
1966 writer.commit()?;
1967 }
1968
1969 Ok(None)
1970 }
1971
1972 fn effective_coverage_block<Provider>(
1981 provider: &Provider,
1982 segment: StaticFileSegment,
1983 highest_static_file_block: BlockNumber,
1984 ) -> ProviderResult<BlockNumber>
1985 where
1986 Provider: PruneCheckpointReader,
1987 {
1988 let Some(prune_segment) = Self::prune_segment_for_static_file(segment) else {
1989 return Ok(highest_static_file_block)
1990 };
1991
1992 let prune_checkpoint_block = provider
1993 .get_prune_checkpoint(prune_segment)?
1994 .and_then(|checkpoint| checkpoint.block_number)
1995 .unwrap_or_default();
1996
1997 Ok(highest_static_file_block.max(prune_checkpoint_block))
1998 }
1999
2000 const fn prune_segment_for_static_file(segment: StaticFileSegment) -> Option<PruneSegment> {
2003 match segment {
2004 StaticFileSegment::Receipts => Some(PruneSegment::Receipts),
2005 StaticFileSegment::TransactionSenders => Some(PruneSegment::SenderRecovery),
2006 StaticFileSegment::AccountChangeSets => Some(PruneSegment::AccountHistory),
2007 StaticFileSegment::StorageChangeSets => Some(PruneSegment::StorageHistory),
2008 StaticFileSegment::Headers | StaticFileSegment::Transactions => None,
2009 }
2010 }
2011
2012 pub fn earliest_history_height(&self) -> BlockNumber {
2020 self.earliest_history_height.load(std::sync::atomic::Ordering::Relaxed)
2021 }
2022
2023 pub(crate) fn set_earliest_history_height(&self, block: BlockNumber) {
2025 self.earliest_history_height.store(block, std::sync::atomic::Ordering::Relaxed);
2026 }
2027
2028 pub fn get_lowest_range(&self, segment: StaticFileSegment) -> Option<SegmentRangeInclusive> {
2032 self.indexes.read().get(segment).and_then(|index| index.min_block_range)
2033 }
2034
2035 pub fn get_lowest_range_start(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
2041 self.get_lowest_range(segment).map(|range| range.start())
2042 }
2043
2044 pub fn get_lowest_range_end(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
2050 self.get_lowest_range(segment).map(|range| range.end())
2051 }
2052
2053 pub fn get_highest_static_file_block(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
2057 self.indexes.read().get(segment).map(|index| index.max_block)
2058 }
2059
2060 fn bound_range(
2067 &self,
2068 range: impl RangeBounds<BlockNumber>,
2069 segment: StaticFileSegment,
2070 ) -> RangeInclusive<BlockNumber> {
2071 let highest_block = self.get_highest_static_file_block(segment).unwrap_or(0);
2072
2073 let start = match range.start_bound() {
2074 Bound::Included(&n) => n,
2075 Bound::Excluded(&n) => n.saturating_add(1),
2076 Bound::Unbounded => 0,
2077 };
2078 let end = match range.end_bound() {
2079 Bound::Included(&n) => n.min(highest_block),
2080 Bound::Excluded(&n) => n.saturating_sub(1).min(highest_block),
2081 Bound::Unbounded => highest_block,
2082 };
2083
2084 start..=end
2085 }
2086
2087 pub fn get_highest_static_file_tx(&self, segment: StaticFileSegment) -> Option<TxNumber> {
2091 self.indexes
2092 .read()
2093 .get(segment)
2094 .and_then(|index| index.available_block_ranges_by_max_tx.as_ref())
2095 .and_then(|index| index.last_key_value().map(|(last_tx, _)| *last_tx))
2096 }
2097
2098 pub fn get_highest_static_files(&self) -> HighestStaticFiles {
2100 HighestStaticFiles {
2101 receipts: self.get_highest_static_file_block(StaticFileSegment::Receipts),
2102 }
2103 }
2104
2105 pub fn find_static_file<T>(
2108 &self,
2109 segment: StaticFileSegment,
2110 func: impl Fn(StaticFileJarProvider<'_, N>) -> ProviderResult<Option<T>>,
2111 ) -> ProviderResult<Option<T>> {
2112 if let Some(ranges) =
2113 self.indexes.read().get(segment).map(|index| &index.expected_block_ranges_by_max_block)
2114 {
2115 for range in ranges.values().rev() {
2117 if let Some(res) = func(self.get_or_create_jar_provider(segment, range)?)? {
2118 return Ok(Some(res));
2119 }
2120 }
2121 }
2122
2123 Ok(None)
2124 }
2125
2126 pub fn fetch_range_with_predicate<T, F, P>(
2132 &self,
2133 segment: StaticFileSegment,
2134 range: Range<u64>,
2135 mut get_fn: F,
2136 mut predicate: P,
2137 ) -> ProviderResult<Vec<T>>
2138 where
2139 F: FnMut(&mut StaticFileCursor<'_>, u64) -> ProviderResult<Option<T>>,
2140 P: FnMut(&T) -> bool,
2141 {
2142 let mut result = Vec::with_capacity((range.end - range.start).min(100) as usize);
2143
2144 macro_rules! get_provider {
2148 ($number:expr) => {{
2149 match self.get_segment_provider(segment, $number) {
2150 Ok(provider) => provider,
2151 Err(
2152 ProviderError::MissingStaticFileBlock(_, _) |
2153 ProviderError::MissingStaticFileTx(_, _),
2154 ) => return Ok(result),
2155 Err(err) => return Err(err),
2156 }
2157 }};
2158 }
2159
2160 let mut provider = get_provider!(range.start);
2161 let mut cursor = provider.cursor()?;
2162
2163 'outer: for number in range {
2165 let mut retrying = false;
2169
2170 'inner: loop {
2172 match get_fn(&mut cursor, number)? {
2173 Some(res) => {
2174 if !predicate(&res) {
2175 break 'outer;
2176 }
2177 result.push(res);
2178 break 'inner;
2179 }
2180 None => {
2181 if retrying {
2182 return Ok(result);
2183 }
2184 drop(cursor);
2189 drop(provider);
2190 provider = get_provider!(number);
2191 cursor = provider.cursor()?;
2192 retrying = true;
2193 }
2194 }
2195 }
2196 }
2197
2198 result.shrink_to_fit();
2199
2200 Ok(result)
2201 }
2202
2203 pub fn fetch_range_iter<'a, T, F>(
2208 &'a self,
2209 segment: StaticFileSegment,
2210 range: Range<u64>,
2211 get_fn: F,
2212 ) -> ProviderResult<impl Iterator<Item = ProviderResult<Option<T>>> + 'a>
2213 where
2214 F: Fn(&mut StaticFileCursor<'_>, u64) -> ProviderResult<Option<T>> + 'a,
2215 T: std::fmt::Debug,
2216 {
2217 let mut provider = self.get_maybe_segment_provider(segment, range.start)?;
2218 Ok(range.map(move |number| {
2219 match provider
2220 .as_ref()
2221 .map(|provider| get_fn(&mut provider.cursor()?, number))
2222 .and_then(|result| result.transpose())
2223 {
2224 Some(result) => result.map(Some),
2225 None => {
2226 provider.take();
2230 provider = self.get_maybe_segment_provider(segment, number)?;
2231 provider
2232 .as_ref()
2233 .map(|provider| get_fn(&mut provider.cursor()?, number))
2234 .and_then(|result| result.transpose())
2235 .transpose()
2236 }
2237 }
2238 }))
2239 }
2240
2241 pub fn directory(&self) -> &Path {
2243 &self.path
2244 }
2245
2246 pub fn get_with_static_file_or_database<T, FS, FD>(
2256 &self,
2257 segment: StaticFileSegment,
2258 number: u64,
2259 fetch_from_static_file: FS,
2260 fetch_from_database: FD,
2261 ) -> ProviderResult<Option<T>>
2262 where
2263 FS: Fn(&Self) -> ProviderResult<Option<T>>,
2264 FD: Fn() -> ProviderResult<Option<T>>,
2265 {
2266 let static_file_upper_bound = if segment.is_block_or_change_based() {
2268 self.get_highest_static_file_block(segment)
2269 } else {
2270 self.get_highest_static_file_tx(segment)
2271 };
2272
2273 if static_file_upper_bound
2274 .is_some_and(|static_file_upper_bound| static_file_upper_bound >= number)
2275 {
2276 return fetch_from_static_file(self);
2277 }
2278 fetch_from_database()
2279 }
2280
2281 pub fn get_range_with_static_file_or_database<T, P, FS, FD>(
2293 &self,
2294 segment: StaticFileSegment,
2295 mut block_or_tx_range: Range<u64>,
2296 fetch_from_static_file: FS,
2297 mut fetch_from_database: FD,
2298 mut predicate: P,
2299 ) -> ProviderResult<Vec<T>>
2300 where
2301 FS: Fn(&Self, Range<u64>, &mut P) -> ProviderResult<Vec<T>>,
2302 FD: FnMut(Range<u64>, P) -> ProviderResult<Vec<T>>,
2303 P: FnMut(&T) -> bool,
2304 {
2305 let mut data = Vec::new();
2306
2307 if let Some(static_file_upper_bound) = if segment.is_block_or_change_based() {
2309 self.get_highest_static_file_block(segment)
2310 } else {
2311 self.get_highest_static_file_tx(segment)
2312 } && block_or_tx_range.start <= static_file_upper_bound
2313 {
2314 let end = block_or_tx_range.end.min(static_file_upper_bound + 1);
2315 data.extend(fetch_from_static_file(
2316 self,
2317 block_or_tx_range.start..end,
2318 &mut predicate,
2319 )?);
2320 block_or_tx_range.start = end;
2321 }
2322
2323 if block_or_tx_range.end > block_or_tx_range.start {
2324 data.extend(fetch_from_database(block_or_tx_range, predicate)?)
2325 }
2326
2327 Ok(data)
2328 }
2329
2330 #[cfg(any(test, feature = "test-utils"))]
2332 pub fn path(&self) -> &Path {
2333 &self.path
2334 }
2335
2336 #[cfg(any(test, feature = "test-utils"))]
2338 pub fn tx_index(&self, segment: StaticFileSegment) -> Option<SegmentRanges> {
2339 self.indexes
2340 .read()
2341 .get(segment)
2342 .and_then(|index| index.available_block_ranges_by_max_tx.as_ref())
2343 .cloned()
2344 }
2345
2346 #[cfg(any(test, feature = "test-utils"))]
2348 pub fn expected_block_index(&self, segment: StaticFileSegment) -> Option<SegmentRanges> {
2349 self.indexes
2350 .read()
2351 .get(segment)
2352 .map(|index| &index.expected_block_ranges_by_max_block)
2353 .cloned()
2354 }
2355}
2356
2357#[derive(Debug)]
2358struct StaticFileSegmentIndex {
2359 min_block_range: Option<SegmentRangeInclusive>,
2371 max_block: u64,
2373 expected_block_ranges_by_max_block: SegmentRanges,
2379 available_block_ranges_by_max_tx: Option<SegmentRanges>,
2386}
2387
2388pub trait StaticFileWriter {
2390 type Primitives: Send + Sync + 'static;
2392
2393 fn get_writer(
2395 &self,
2396 block: BlockNumber,
2397 segment: StaticFileSegment,
2398 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2399
2400 fn latest_writer(
2403 &self,
2404 segment: StaticFileSegment,
2405 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2406
2407 fn commit(&self) -> ProviderResult<()>;
2409
2410 fn has_unwind_queued(&self) -> bool;
2412
2413 fn finalize(&self) -> ProviderResult<()>;
2417}
2418
2419impl<N: NodePrimitives> StaticFileWriter for StaticFileProvider<N> {
2420 type Primitives = N;
2421
2422 fn get_writer(
2423 &self,
2424 block: BlockNumber,
2425 segment: StaticFileSegment,
2426 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
2427 if self.access.is_read_only() {
2428 return Err(ProviderError::ReadOnlyStaticFileAccess);
2429 }
2430
2431 trace!(target: "providers::static_file", ?block, ?segment, "Getting static file writer.");
2432 self.writers.get_or_create(segment, || {
2433 StaticFileProviderRW::new(segment, block, Arc::downgrade(&self.0), self.metrics.clone())
2434 })
2435 }
2436
2437 fn latest_writer(
2438 &self,
2439 segment: StaticFileSegment,
2440 ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
2441 let genesis_number = self.0.as_ref().genesis_block_number();
2442 self.get_writer(
2443 self.get_highest_static_file_block(segment).unwrap_or(genesis_number),
2444 segment,
2445 )
2446 }
2447
2448 fn commit(&self) -> ProviderResult<()> {
2449 self.writers.commit()
2450 }
2451
2452 fn has_unwind_queued(&self) -> bool {
2453 self.writers.has_unwind_queued()
2454 }
2455
2456 fn finalize(&self) -> ProviderResult<()> {
2457 self.writers.finalize()
2458 }
2459}
2460
2461impl<N: NodePrimitives> ChangeSetReader for StaticFileProvider<N> {
2462 fn account_block_changeset(
2463 &self,
2464 block_number: BlockNumber,
2465 ) -> ProviderResult<Vec<reth_db::models::AccountBeforeTx>> {
2466 let provider = match self.get_segment_provider_for_block(
2467 StaticFileSegment::AccountChangeSets,
2468 block_number,
2469 None,
2470 ) {
2471 Ok(provider) => provider,
2472 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(Vec::new()),
2473 Err(err) => return Err(err),
2474 };
2475
2476 if let Some(offset) = provider.read_changeset_offset(block_number)? {
2477 let mut cursor = provider.cursor()?;
2478 let mut changeset = Vec::with_capacity(offset.num_changes() as usize);
2479
2480 for i in offset.changeset_range() {
2481 if let Some(change) =
2482 cursor.get_one::<reth_db::static_file::AccountChangesetMask>(i.into())?
2483 {
2484 changeset.push(change)
2485 }
2486 }
2487 Ok(changeset)
2488 } else {
2489 Ok(Vec::new())
2490 }
2491 }
2492
2493 fn get_account_before_block(
2494 &self,
2495 block_number: BlockNumber,
2496 address: Address,
2497 ) -> ProviderResult<Option<reth_db::models::AccountBeforeTx>> {
2498 let provider = match self.get_segment_provider_for_block(
2499 StaticFileSegment::AccountChangeSets,
2500 block_number,
2501 None,
2502 ) {
2503 Ok(provider) => provider,
2504 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(None),
2505 Err(err) => return Err(err),
2506 };
2507
2508 let Some(offset) = provider.read_changeset_offset(block_number)? else {
2509 return Ok(None);
2510 };
2511
2512 let mut cursor = provider.cursor()?;
2513 let range = offset.changeset_range();
2514 let mut low = range.start;
2515 let mut high = range.end;
2516
2517 while low < high {
2518 let mid = low + (high - low) / 2;
2519 if let Some(change) =
2520 cursor.get_one::<reth_db::static_file::AccountChangesetMask>(mid.into())?
2521 {
2522 if change.address < address {
2523 low = mid + 1;
2524 } else {
2525 high = mid;
2526 }
2527 } else {
2528 debug!(
2531 target: "providers::static_file",
2532 ?low,
2533 ?mid,
2534 ?high,
2535 ?range,
2536 ?block_number,
2537 ?address,
2538 "Cannot continue binary search for account changeset fetch"
2539 );
2540 low = range.end;
2541 break;
2542 }
2543 }
2544
2545 if low < range.end &&
2546 let Some(change) = cursor
2547 .get_one::<reth_db::static_file::AccountChangesetMask>(low.into())?
2548 .filter(|change| change.address == address)
2549 {
2550 return Ok(Some(change));
2551 }
2552
2553 Ok(None)
2554 }
2555
2556 fn account_changesets_range(
2557 &self,
2558 range: impl core::ops::RangeBounds<BlockNumber>,
2559 ) -> ProviderResult<Vec<(BlockNumber, reth_db::models::AccountBeforeTx)>> {
2560 let range = self.bound_range(range, StaticFileSegment::AccountChangeSets);
2561 self.walk_account_changeset_range(range).collect()
2562 }
2563}
2564
2565impl<N: NodePrimitives> StorageChangeSetReader for StaticFileProvider<N> {
2566 fn storage_changeset(
2567 &self,
2568 block_number: BlockNumber,
2569 ) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
2570 let provider = match self.get_segment_provider_for_block(
2571 StaticFileSegment::StorageChangeSets,
2572 block_number,
2573 None,
2574 ) {
2575 Ok(provider) => provider,
2576 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(Vec::new()),
2577 Err(err) => return Err(err),
2578 };
2579
2580 if let Some(offset) = provider.read_changeset_offset(block_number)? {
2581 let mut cursor = provider.cursor()?;
2582 let mut changeset = Vec::with_capacity(offset.num_changes() as usize);
2583
2584 for i in offset.changeset_range() {
2585 if let Some(change) = cursor.get_one::<StorageChangesetMask>(i.into())? {
2586 let block_address = BlockNumberAddress((block_number, change.address));
2587 let entry = StorageEntry { key: change.key, value: change.value };
2588 changeset.push((block_address, entry));
2589 }
2590 }
2591 Ok(changeset)
2592 } else {
2593 Ok(Vec::new())
2594 }
2595 }
2596
2597 fn get_storage_before_block(
2598 &self,
2599 block_number: BlockNumber,
2600 address: Address,
2601 storage_key: B256,
2602 ) -> ProviderResult<Option<StorageEntry>> {
2603 let provider = match self.get_segment_provider_for_block(
2604 StaticFileSegment::StorageChangeSets,
2605 block_number,
2606 None,
2607 ) {
2608 Ok(provider) => provider,
2609 Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(None),
2610 Err(err) => return Err(err),
2611 };
2612
2613 let Some(offset) = provider.read_changeset_offset(block_number)? else {
2614 return Ok(None);
2615 };
2616
2617 let mut cursor = provider.cursor()?;
2618 let range = offset.changeset_range();
2619 let mut low = range.start;
2620 let mut high = range.end;
2621
2622 while low < high {
2623 let mid = low + (high - low) / 2;
2624 if let Some(change) = cursor.get_one::<StorageChangesetMask>(mid.into())? {
2625 match (change.address, change.key).cmp(&(address, storage_key)) {
2626 std::cmp::Ordering::Less => low = mid + 1,
2627 _ => high = mid,
2628 }
2629 } else {
2630 debug!(
2631 target: "providers::static_file",
2632 ?low,
2633 ?mid,
2634 ?high,
2635 ?range,
2636 ?block_number,
2637 ?address,
2638 ?storage_key,
2639 "Cannot continue binary search for storage changeset fetch"
2640 );
2641 low = range.end;
2642 break;
2643 }
2644 }
2645
2646 if low < range.end &&
2647 let Some(change) = cursor
2648 .get_one::<StorageChangesetMask>(low.into())?
2649 .filter(|change| change.address == address && change.key == storage_key)
2650 {
2651 return Ok(Some(StorageEntry { key: change.key, value: change.value }));
2652 }
2653
2654 Ok(None)
2655 }
2656
2657 fn storage_changesets_range(
2658 &self,
2659 range: impl RangeBounds<BlockNumber>,
2660 ) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
2661 let range = self.bound_range(range, StaticFileSegment::StorageChangeSets);
2662 self.walk_storage_changeset_range(range).collect()
2663 }
2664}
2665
2666impl<N: NodePrimitives> StaticFileProvider<N> {
2667 pub fn walk_account_changeset_range(
2677 &self,
2678 range: impl RangeBounds<BlockNumber>,
2679 ) -> StaticFileAccountChangesetWalker<Self> {
2680 StaticFileAccountChangesetWalker::new(self.clone(), range)
2681 }
2682
2683 pub fn walk_storage_changeset_range(
2685 &self,
2686 range: impl RangeBounds<BlockNumber>,
2687 ) -> StaticFileStorageChangesetWalker<Self> {
2688 StaticFileStorageChangesetWalker::new(self.clone(), range)
2689 }
2690}
2691
2692impl<N: NodePrimitives<BlockHeader: Value>> HeaderProvider for StaticFileProvider<N> {
2693 type Header = N::BlockHeader;
2694
2695 fn header(&self, block_hash: BlockHash) -> ProviderResult<Option<Self::Header>> {
2696 self.find_static_file(StaticFileSegment::Headers, |jar_provider| {
2697 Ok(jar_provider
2698 .cursor()?
2699 .get_two::<HeaderWithHashMask<Self::Header>>((&block_hash).into())?
2700 .and_then(|(header, hash)| {
2701 if hash == block_hash {
2702 return Some(header);
2703 }
2704 None
2705 }))
2706 })
2707 }
2708
2709 fn header_by_number(&self, num: BlockNumber) -> ProviderResult<Option<Self::Header>> {
2710 self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2711 .and_then(|provider| provider.header_by_number(num))
2712 .or_else(|err| {
2713 if let ProviderError::MissingStaticFileBlock(_, _) = err {
2714 Ok(None)
2715 } else {
2716 Err(err)
2717 }
2718 })
2719 }
2720
2721 fn headers_range(
2722 &self,
2723 range: impl RangeBounds<BlockNumber>,
2724 ) -> ProviderResult<Vec<Self::Header>> {
2725 self.fetch_range_with_predicate(
2726 StaticFileSegment::Headers,
2727 to_range(range),
2728 |cursor, number| cursor.get_one::<HeaderMask<Self::Header>>(number.into()),
2729 |_| true,
2730 )
2731 }
2732
2733 fn sealed_header(
2734 &self,
2735 num: BlockNumber,
2736 ) -> ProviderResult<Option<SealedHeader<Self::Header>>> {
2737 self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2738 .and_then(|provider| provider.sealed_header(num))
2739 .or_else(|err| {
2740 if let ProviderError::MissingStaticFileBlock(_, _) = err {
2741 Ok(None)
2742 } else {
2743 Err(err)
2744 }
2745 })
2746 }
2747
2748 fn sealed_headers_while(
2749 &self,
2750 range: impl RangeBounds<BlockNumber>,
2751 predicate: impl FnMut(&SealedHeader<Self::Header>) -> bool,
2752 ) -> ProviderResult<Vec<SealedHeader<Self::Header>>> {
2753 self.fetch_range_with_predicate(
2754 StaticFileSegment::Headers,
2755 to_range(range),
2756 |cursor, number| {
2757 Ok(cursor
2758 .get_two::<HeaderWithHashMask<Self::Header>>(number.into())?
2759 .map(|(header, hash)| SealedHeader::new(header, hash)))
2760 },
2761 predicate,
2762 )
2763 }
2764}
2765
2766impl<N: NodePrimitives> BlockHashReader for StaticFileProvider<N> {
2767 fn block_hash(&self, num: u64) -> ProviderResult<Option<B256>> {
2768 self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2769 .and_then(|provider| provider.block_hash(num))
2770 .or_else(|err| {
2771 if let ProviderError::MissingStaticFileBlock(_, _) = err {
2772 Ok(None)
2773 } else {
2774 Err(err)
2775 }
2776 })
2777 }
2778
2779 fn canonical_hashes_range(
2780 &self,
2781 start: BlockNumber,
2782 end: BlockNumber,
2783 ) -> ProviderResult<Vec<B256>> {
2784 self.fetch_range_with_predicate(
2785 StaticFileSegment::Headers,
2786 start..end,
2787 |cursor, number| cursor.get_one::<BlockHashMask>(number.into()),
2788 |_| true,
2789 )
2790 }
2791}
2792
2793impl<N: NodePrimitives<SignedTx: Value + SignedTransaction, Receipt: Value>> ReceiptProvider
2794 for StaticFileProvider<N>
2795{
2796 type Receipt = N::Receipt;
2797
2798 fn receipt(&self, num: TxNumber) -> ProviderResult<Option<Self::Receipt>> {
2799 self.get_segment_provider_for_transaction(StaticFileSegment::Receipts, num, None)
2800 .and_then(|provider| provider.receipt(num))
2801 .or_else(|err| {
2802 if let ProviderError::MissingStaticFileTx(_, _) = err {
2803 Ok(None)
2804 } else {
2805 Err(err)
2806 }
2807 })
2808 }
2809
2810 fn receipt_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Receipt>> {
2811 if let Some(num) = self.transaction_id(hash)? {
2812 return self.receipt(num);
2813 }
2814 Ok(None)
2815 }
2816
2817 fn receipts_by_block(
2818 &self,
2819 _block: BlockHashOrNumber,
2820 ) -> ProviderResult<Option<Vec<Self::Receipt>>> {
2821 unreachable!()
2822 }
2823
2824 fn receipts_by_tx_range(
2825 &self,
2826 range: impl RangeBounds<TxNumber>,
2827 ) -> ProviderResult<Vec<Self::Receipt>> {
2828 self.fetch_range_with_predicate(
2829 StaticFileSegment::Receipts,
2830 to_range(range),
2831 |cursor, number| cursor.get_one::<ReceiptMask<Self::Receipt>>(number.into()),
2832 |_| true,
2833 )
2834 }
2835
2836 fn receipts_by_block_range(
2837 &self,
2838 _block_range: RangeInclusive<BlockNumber>,
2839 ) -> ProviderResult<Vec<Vec<Self::Receipt>>> {
2840 Err(ProviderError::UnsupportedProvider)
2841 }
2842}
2843
2844impl<N: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>> TransactionsProviderExt
2845 for StaticFileProvider<N>
2846{
2847 fn transaction_hashes_by_range(
2848 &self,
2849 tx_range: Range<TxNumber>,
2850 ) -> ProviderResult<Vec<(TxHash, TxNumber)>> {
2851 let tx_range_size = (tx_range.end - tx_range.start) as usize;
2852
2853 let chunk_size = 100;
2857
2858 let chunks = tx_range
2860 .clone()
2861 .step_by(chunk_size)
2862 .map(|start| start..std::cmp::min(start + chunk_size as u64, tx_range.end));
2863 let mut channels = Vec::with_capacity(tx_range_size.div_ceil(chunk_size));
2864
2865 for chunk_range in chunks {
2866 let (channel_tx, channel_rx) = mpsc::channel();
2867 channels.push(channel_rx);
2868
2869 let manager = self.clone();
2870
2871 rayon::spawn(move || {
2874 let _ = manager.fetch_range_with_predicate(
2875 StaticFileSegment::Transactions,
2876 chunk_range,
2877 |cursor, number| {
2878 Ok(cursor
2879 .get_one::<TransactionMask<Self::Transaction>>(number.into())?
2880 .map(|transaction| {
2881 let _ = channel_tx.send(transaction_hash((number, transaction)));
2882 }))
2883 },
2884 |_| true,
2885 );
2886 });
2887 }
2888
2889 let mut tx_list = Vec::with_capacity(tx_range_size);
2890
2891 for channel in channels {
2893 while let Ok(tx) = channel.recv() {
2894 let (tx_hash, tx_id) = tx.map_err(|boxed| *boxed)?;
2895 tx_list.push((tx_hash, tx_id));
2896 }
2897 }
2898
2899 Ok(tx_list)
2900 }
2901}
2902
2903impl<N: NodePrimitives<SignedTx: Decompress + SignedTransaction>> TransactionsProvider
2904 for StaticFileProvider<N>
2905{
2906 type Transaction = N::SignedTx;
2907
2908 fn transaction_id(&self, tx_hash: TxHash) -> ProviderResult<Option<TxNumber>> {
2909 self.find_static_file(StaticFileSegment::Transactions, |jar_provider| {
2910 let mut cursor = jar_provider.cursor()?;
2911 if cursor
2912 .get_one::<TransactionMask<Self::Transaction>>((&tx_hash).into())?
2913 .and_then(|tx| (*tx.tx_hash() == tx_hash).then_some(tx))
2914 .is_some()
2915 {
2916 Ok(cursor.number())
2917 } else {
2918 Ok(None)
2919 }
2920 })
2921 }
2922
2923 fn transaction_by_id(&self, num: TxNumber) -> ProviderResult<Option<Self::Transaction>> {
2924 self.get_segment_provider_for_transaction(StaticFileSegment::Transactions, num, None)
2925 .and_then(|provider| provider.transaction_by_id(num))
2926 .or_else(|err| {
2927 if let ProviderError::MissingStaticFileTx(_, _) = err {
2928 Ok(None)
2929 } else {
2930 Err(err)
2931 }
2932 })
2933 }
2934
2935 fn transaction_by_id_unhashed(
2936 &self,
2937 num: TxNumber,
2938 ) -> ProviderResult<Option<Self::Transaction>> {
2939 self.get_segment_provider_for_transaction(StaticFileSegment::Transactions, num, None)
2940 .and_then(|provider| provider.transaction_by_id_unhashed(num))
2941 .or_else(|err| {
2942 if let ProviderError::MissingStaticFileTx(_, _) = err {
2943 Ok(None)
2944 } else {
2945 Err(err)
2946 }
2947 })
2948 }
2949
2950 fn transaction_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Transaction>> {
2951 self.find_static_file(StaticFileSegment::Transactions, |jar_provider| {
2952 Ok(jar_provider
2953 .cursor()?
2954 .get_one::<TransactionMask<Self::Transaction>>((&hash).into())?
2955 .and_then(|tx| (*tx.tx_hash() == hash).then_some(tx)))
2956 })
2957 }
2958
2959 fn transaction_by_hash_with_meta(
2960 &self,
2961 _hash: TxHash,
2962 ) -> ProviderResult<Option<(Self::Transaction, TransactionMeta)>> {
2963 Err(ProviderError::UnsupportedProvider)
2965 }
2966
2967 fn transactions_by_block(
2968 &self,
2969 _block_id: BlockHashOrNumber,
2970 ) -> ProviderResult<Option<Vec<Self::Transaction>>> {
2971 Err(ProviderError::UnsupportedProvider)
2973 }
2974
2975 fn transactions_by_block_range(
2976 &self,
2977 _range: impl RangeBounds<BlockNumber>,
2978 ) -> ProviderResult<Vec<Vec<Self::Transaction>>> {
2979 Err(ProviderError::UnsupportedProvider)
2981 }
2982
2983 fn transactions_by_tx_range(
2984 &self,
2985 range: impl RangeBounds<TxNumber>,
2986 ) -> ProviderResult<Vec<Self::Transaction>> {
2987 self.fetch_range_with_predicate(
2988 StaticFileSegment::Transactions,
2989 to_range(range),
2990 |cursor, number| cursor.get_one::<TransactionMask<Self::Transaction>>(number.into()),
2991 |_| true,
2992 )
2993 }
2994
2995 fn senders_by_tx_range(
2996 &self,
2997 range: impl RangeBounds<TxNumber>,
2998 ) -> ProviderResult<Vec<Address>> {
2999 self.fetch_range_with_predicate(
3000 StaticFileSegment::TransactionSenders,
3001 to_range(range),
3002 |cursor, number| cursor.get_one::<TransactionSenderMask>(number.into()),
3003 |_| true,
3004 )
3005 }
3006
3007 fn transaction_sender(&self, id: TxNumber) -> ProviderResult<Option<Address>> {
3008 self.get_segment_provider_for_transaction(StaticFileSegment::TransactionSenders, id, None)
3009 .and_then(|provider| provider.transaction_sender(id))
3010 .or_else(|err| {
3011 if let ProviderError::MissingStaticFileTx(_, _) = err {
3012 Ok(None)
3013 } else {
3014 Err(err)
3015 }
3016 })
3017 }
3018}
3019
3020impl<N: NodePrimitives> BlockNumReader for StaticFileProvider<N> {
3021 fn chain_info(&self) -> ProviderResult<ChainInfo> {
3022 Err(ProviderError::UnsupportedProvider)
3024 }
3025
3026 fn best_block_number(&self) -> ProviderResult<BlockNumber> {
3027 Err(ProviderError::UnsupportedProvider)
3029 }
3030
3031 fn last_block_number(&self) -> ProviderResult<BlockNumber> {
3032 Ok(self.get_highest_static_file_block(StaticFileSegment::Headers).unwrap_or_default())
3033 }
3034
3035 fn block_number(&self, _hash: B256) -> ProviderResult<Option<BlockNumber>> {
3036 Err(ProviderError::UnsupportedProvider)
3038 }
3039}
3040
3041impl<N: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>> BlockReader
3044 for StaticFileProvider<N>
3045{
3046 type Block = N::Block;
3047
3048 fn find_block_by_hash(
3049 &self,
3050 _hash: B256,
3051 _source: BlockSource,
3052 ) -> ProviderResult<Option<Self::Block>> {
3053 Err(ProviderError::UnsupportedProvider)
3055 }
3056
3057 fn block(&self, _id: BlockHashOrNumber) -> ProviderResult<Option<Self::Block>> {
3058 Err(ProviderError::UnsupportedProvider)
3060 }
3061
3062 fn pending_block(&self) -> ProviderResult<Option<Arc<RecoveredBlock<Self::Block>>>> {
3063 Err(ProviderError::UnsupportedProvider)
3065 }
3066
3067 fn pending_block_and_receipts(
3068 &self,
3069 ) -> ProviderResult<Option<RecoveredBlockAndExecutionOutput<Self::Block, Self::Receipt>>> {
3070 Err(ProviderError::UnsupportedProvider)
3072 }
3073
3074 fn recovered_block(
3075 &self,
3076 _id: BlockHashOrNumber,
3077 _transaction_kind: TransactionVariant,
3078 ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
3079 Err(ProviderError::UnsupportedProvider)
3081 }
3082
3083 fn sealed_block_with_senders(
3084 &self,
3085 _id: BlockHashOrNumber,
3086 _transaction_kind: TransactionVariant,
3087 ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
3088 Err(ProviderError::UnsupportedProvider)
3090 }
3091
3092 fn block_range(&self, _range: RangeInclusive<BlockNumber>) -> ProviderResult<Vec<Self::Block>> {
3093 Err(ProviderError::UnsupportedProvider)
3095 }
3096
3097 fn block_with_senders_range(
3098 &self,
3099 _range: RangeInclusive<BlockNumber>,
3100 ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
3101 Err(ProviderError::UnsupportedProvider)
3102 }
3103
3104 fn recovered_block_range(
3105 &self,
3106 _range: RangeInclusive<BlockNumber>,
3107 ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
3108 Err(ProviderError::UnsupportedProvider)
3109 }
3110
3111 fn block_by_transaction_id(&self, _id: TxNumber) -> ProviderResult<Option<BlockNumber>> {
3112 Err(ProviderError::UnsupportedProvider)
3113 }
3114}
3115
3116impl<N: NodePrimitives> BlockBodyIndicesProvider for StaticFileProvider<N> {
3117 fn block_body_indices(&self, _num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
3118 Err(ProviderError::UnsupportedProvider)
3119 }
3120
3121 fn block_body_indices_range(
3122 &self,
3123 _range: RangeInclusive<BlockNumber>,
3124 ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
3125 Err(ProviderError::UnsupportedProvider)
3126 }
3127}
3128
3129impl<N: NodePrimitives> StatsReader for StaticFileProvider<N> {
3130 fn count_entries<T: Table>(&self) -> ProviderResult<usize> {
3131 match T::NAME {
3132 tables::CanonicalHeaders::NAME |
3133 tables::Headers::<Header>::NAME |
3134 tables::HeaderTerminalDifficulties::NAME => Ok(self
3135 .get_highest_static_file_block(StaticFileSegment::Headers)
3136 .map(|block| block + 1)
3137 .unwrap_or_default()
3138 as usize),
3139 tables::Receipts::<Receipt>::NAME => Ok(self
3140 .get_highest_static_file_tx(StaticFileSegment::Receipts)
3141 .map(|receipts| receipts + 1)
3142 .unwrap_or_default() as usize),
3143 tables::Transactions::<TransactionSigned>::NAME => Ok(self
3144 .get_highest_static_file_tx(StaticFileSegment::Transactions)
3145 .map(|txs| txs + 1)
3146 .unwrap_or_default()
3147 as usize),
3148 tables::TransactionSenders::NAME => Ok(self
3149 .get_highest_static_file_tx(StaticFileSegment::TransactionSenders)
3150 .map(|txs| txs + 1)
3151 .unwrap_or_default() as usize),
3152 _ => Err(ProviderError::UnsupportedProvider),
3153 }
3154 }
3155}
3156
3157#[inline]
3159fn transaction_hash<T>(entry: (TxNumber, T)) -> Result<(B256, TxNumber), Box<ProviderError>>
3160where
3161 T: TxHashRef,
3162{
3163 let (tx_id, tx) = entry;
3164 Ok((*tx.tx_hash(), tx_id))
3165}
3166
3167#[cfg(test)]
3168mod tests {
3169 use std::{
3170 collections::BTreeMap,
3171 sync::{atomic::Ordering, mpsc},
3172 thread,
3173 time::{Duration, Instant},
3174 };
3175
3176 use alloy_consensus::Header;
3177 use alloy_primitives::B256;
3178 use reth_chain_state::EthPrimitives;
3179 use reth_db::test_utils::create_test_static_files_dir;
3180 use reth_static_file_types::{SegmentRangeInclusive, StaticFileSegment};
3181
3182 use super::StaticFileWriter;
3183 use crate::{providers::StaticFileProvider, BlockHashReader, StaticFileProviderBuilder};
3184
3185 #[test]
3186 fn recovery_prunes_transaction_zero_at_empty_genesis() -> eyre::Result<()> {
3187 use crate::{
3188 test_utils::create_test_provider_factory, StaticFileProviderFactory,
3189 TransactionsProvider,
3190 };
3191 use alloy_consensus::{SignableTransaction, TxLegacy};
3192 use alloy_primitives::Signature;
3193 use reth_db::{tables, transaction::DbTxMut};
3194
3195 let factory = create_test_provider_factory();
3196 let provider = factory.provider_rw()?;
3197 provider.tx_ref().put::<tables::BlockBodyIndices>(0, Default::default())?;
3198 provider.commit()?;
3199
3200 let static_files = factory.static_file_provider();
3201 let segment = StaticFileSegment::Transactions;
3202 let tx = TxLegacy::default().into_signed(Signature::test_signature()).into();
3203 {
3204 let mut writer = static_files.latest_writer(segment)?;
3205 writer.increment_block(0)?;
3206 writer.increment_block(1)?;
3207 writer.append_transaction(0, &tx)?;
3208 writer.commit()?;
3209 }
3210 assert!(static_files.transaction_by_id(0)?.is_some());
3211
3212 assert_eq!(static_files.check_consistency(&factory.provider()?)?, None);
3213 assert_eq!(static_files.get_highest_static_file_block(segment), Some(0));
3214 assert!(static_files.transaction_by_id(0)?.is_none());
3215
3216 let mut writer = static_files.latest_writer(segment)?;
3217 writer.increment_block(1)?;
3218 writer.append_transaction(0, &tx)?;
3219 writer.commit()?;
3220 Ok(())
3221 }
3222
3223 #[test]
3224 fn stale_cache_fill_does_not_survive_index_reinitialization() -> eyre::Result<()> {
3225 let (static_dir, _) = create_test_static_files_dir();
3226 let static_files: StaticFileProvider<EthPrimitives> =
3227 StaticFileProviderBuilder::read_write(&static_dir)
3228 .with_blocks_per_file_for_segment(StaticFileSegment::Headers, 10)
3229 .with_blocks_per_file_for_segment(StaticFileSegment::Receipts, 2)
3230 .build()?;
3231
3232 let hash_0 = B256::repeat_byte(0x10);
3233 let header_0 = Header { number: 0, ..Default::default() };
3234 {
3235 let mut writer = static_files.latest_writer(StaticFileSegment::Headers)?;
3236 writer.append_header(&header_0, &hash_0)?;
3237 writer.commit()?;
3238 }
3239
3240 {
3243 let mut writer = static_files.latest_writer(StaticFileSegment::Receipts)?;
3244 for block in 0..=3 {
3245 writer.increment_block(block)?;
3246 }
3247 writer.commit()?;
3248 }
3249 static_files.delete_jar(StaticFileSegment::Receipts, 0)?;
3250
3251 let (loaded_tx, loaded_rx) = mpsc::sync_channel(0);
3252 let (resume_tx, resume_rx) = mpsc::sync_channel(0);
3253 static_files.before_cache_insert.set(move |segment| {
3254 assert_eq!(segment, StaticFileSegment::Headers);
3255 loaded_tx.send(()).expect("test controller dropped");
3256 resume_rx
3257 .recv_timeout(Duration::from_secs(10))
3258 .expect("test controller did not resume");
3259 });
3260
3261 let reader_static_files = static_files.clone();
3263 let reader = thread::spawn(move || reader_static_files.block_hash(0));
3264 loaded_rx
3265 .recv_timeout(Duration::from_secs(10))
3266 .expect("reader did not reach cache-fill hook");
3267
3268 let hash_1 = B256::repeat_byte(0x11);
3271 let header_1 = Header { number: 1, ..Default::default() };
3272 {
3273 let mut writer = static_files.latest_writer(StaticFileSegment::Headers)?;
3274 writer.append_header(&header_1, &hash_1)?;
3275 writer.commit()?;
3276 }
3277 static_files.delete_jar(StaticFileSegment::Receipts, 2)?;
3278
3279 resume_tx.send(()).expect("reader dropped before cache publication");
3280 assert_eq!(reader.join().expect("reader panicked")?, Some(hash_0));
3281 assert_eq!(
3282 static_files.block_hash(1)?,
3283 Some(hash_1),
3284 "cache fill started before reinitialization repopulated an invalidated jar"
3285 );
3286
3287 Ok(())
3288 }
3289
3290 #[test]
3291 fn cache_fill_validation_is_serialized_with_index_reinitialization() -> eyre::Result<()> {
3292 let (static_dir, _) = create_test_static_files_dir();
3293 let static_files: StaticFileProvider<EthPrimitives> =
3294 StaticFileProviderBuilder::read_write(&static_dir).with_blocks_per_file(10).build()?;
3295 let hash = B256::repeat_byte(0x10);
3296 {
3297 let mut writer = static_files.latest_writer(StaticFileSegment::Headers)?;
3298 writer.append_header(&Header::default(), &hash)?;
3299 writer.commit()?;
3300 }
3301 static_files.remove_cached_provider(StaticFileSegment::Headers, 9);
3302
3303 let generation = static_files.cache_generation.load(Ordering::Acquire);
3304 let (validated_tx, validated_rx) = mpsc::sync_channel(0);
3305 let (resume_tx, resume_rx) = mpsc::sync_channel(0);
3306 static_files.after_cache_validation.set(move |segment| {
3307 assert_eq!(segment, StaticFileSegment::Headers);
3308 validated_tx.send(()).expect("test controller dropped");
3309 resume_rx
3310 .recv_timeout(Duration::from_secs(10))
3311 .expect("test controller did not resume");
3312 });
3313
3314 let reader_static_files = static_files.clone();
3315 let reader = thread::spawn(move || reader_static_files.block_hash(0));
3316 validated_rx
3317 .recv_timeout(Duration::from_secs(10))
3318 .expect("reader did not reach cache validation");
3319
3320 assert!(static_files.map.try_get(&(9, StaticFileSegment::Headers)).is_locked());
3323 let invalidator_static_files = static_files.clone();
3324 let invalidator = thread::spawn(move || invalidator_static_files.initialize_index());
3325 let deadline = Instant::now() + Duration::from_secs(10);
3326 while static_files.cache_generation.load(Ordering::Acquire) == generation {
3327 assert!(Instant::now() < deadline, "reinitialization did not invalidate the cache");
3328 thread::sleep(Duration::from_millis(1));
3329 }
3330
3331 resume_tx.send(()).expect("reader dropped before cache publication");
3334 assert_eq!(reader.join().expect("reader panicked")?, Some(hash));
3335 invalidator.join().expect("invalidator panicked")?;
3336 assert!(static_files.map.is_empty());
3337 assert_eq!(static_files.block_hash(0)?, Some(hash));
3338
3339 Ok(())
3340 }
3341
3342 #[test]
3343 fn test_find_fixed_range_with_block_index() -> eyre::Result<()> {
3344 let (static_dir, _) = create_test_static_files_dir();
3345 let sf_rw: StaticFileProvider<EthPrimitives> =
3346 StaticFileProviderBuilder::read_write(&static_dir).with_blocks_per_file(100).build()?;
3347
3348 let segment = StaticFileSegment::Headers;
3349
3350 assert_eq!(
3352 sf_rw.find_fixed_range_with_block_index(segment, None, 0),
3353 SegmentRangeInclusive::new(0, 99)
3354 );
3355 assert_eq!(
3356 sf_rw.find_fixed_range_with_block_index(segment, None, 250),
3357 SegmentRangeInclusive::new(200, 299)
3358 );
3359
3360 assert_eq!(
3362 sf_rw.find_fixed_range_with_block_index(segment, Some(&BTreeMap::new()), 150),
3363 SegmentRangeInclusive::new(100, 199)
3364 );
3365
3366 let block_index = BTreeMap::from_iter([
3368 (99, SegmentRangeInclusive::new(0, 99)),
3369 (199, SegmentRangeInclusive::new(100, 199)),
3370 (299, SegmentRangeInclusive::new(200, 299)),
3371 ]);
3372
3373 assert_eq!(
3375 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 0),
3376 SegmentRangeInclusive::new(0, 99)
3377 );
3378 assert_eq!(
3379 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 50),
3380 SegmentRangeInclusive::new(0, 99)
3381 );
3382 assert_eq!(
3383 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 99),
3384 SegmentRangeInclusive::new(0, 99)
3385 );
3386 assert_eq!(
3387 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 100),
3388 SegmentRangeInclusive::new(100, 199)
3389 );
3390 assert_eq!(
3391 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 150),
3392 SegmentRangeInclusive::new(100, 199)
3393 );
3394 assert_eq!(
3395 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 199),
3396 SegmentRangeInclusive::new(100, 199)
3397 );
3398
3399 assert_eq!(
3402 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 300),
3403 SegmentRangeInclusive::new(300, 399)
3404 );
3405 assert_eq!(
3406 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 350),
3407 SegmentRangeInclusive::new(300, 399)
3408 );
3409
3410 assert_eq!(
3412 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 500),
3413 SegmentRangeInclusive::new(500, 599)
3414 );
3415
3416 assert_eq!(
3418 sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 1000),
3419 SegmentRangeInclusive::new(1000, 1099)
3420 );
3421
3422 let mixed_size_index = BTreeMap::from_iter([
3425 (49, SegmentRangeInclusive::new(0, 49)), (149, SegmentRangeInclusive::new(50, 149)), (349, SegmentRangeInclusive::new(150, 349)), ]);
3429
3430 assert_eq!(
3432 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 25),
3433 SegmentRangeInclusive::new(0, 49)
3434 );
3435 assert_eq!(
3436 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 100),
3437 SegmentRangeInclusive::new(50, 149)
3438 );
3439 assert_eq!(
3440 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 200),
3441 SegmentRangeInclusive::new(150, 349)
3442 );
3443
3444 assert_eq!(
3447 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 350),
3448 SegmentRangeInclusive::new(350, 449)
3449 );
3450 assert_eq!(
3451 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 450),
3452 SegmentRangeInclusive::new(450, 549)
3453 );
3454 assert_eq!(
3455 sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 550),
3456 SegmentRangeInclusive::new(550, 649)
3457 );
3458
3459 Ok(())
3460 }
3461}