Skip to main content

reth_provider/providers/static_file/
manager.rs

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
62/// Alias type for a map that can be queried for block or transaction ranges. It uses `u64` to
63/// represent either a block or a transaction number end of a static file range.
64type SegmentRanges = BTreeMap<u64, SegmentRangeInclusive>;
65
66/// Access mode on a static file provider. RO/RW.
67#[derive(Debug, Default, PartialEq, Eq)]
68pub enum StaticFileAccess {
69    /// Read-only access.
70    #[default]
71    RO,
72    /// Read-write access.
73    RW,
74}
75
76impl StaticFileAccess {
77    /// Returns `true` if read-only access.
78    pub const fn is_read_only(&self) -> bool {
79        matches!(self, Self::RO)
80    }
81
82    /// Returns `true` if read-write access.
83    pub const fn is_read_write(&self) -> bool {
84        matches!(self, Self::RW)
85    }
86}
87
88/// Context for static file block writes.
89///
90/// Contains target segments and pruning configuration.
91#[derive(Debug, Clone, Copy, Default)]
92pub struct StaticFileWriteCtx {
93    /// Whether transaction senders should be written to static files.
94    pub write_senders: bool,
95    /// Whether receipts should be written to static files.
96    pub write_receipts: bool,
97    /// Whether account changesets should be written to static files.
98    pub write_account_changesets: bool,
99    /// Whether storage changesets should be written to static files.
100    pub write_storage_changesets: bool,
101    /// The current chain tip block number (for pruning).
102    pub tip: BlockNumber,
103    /// The prune mode for receipts, if any.
104    pub receipts_prune_mode: Option<reth_prune_types::PruneMode>,
105    /// Whether receipts are prunable (based on storage settings and prune distance).
106    pub receipts_prunable: bool,
107}
108
109/// [`StaticFileProvider`] manages all existing [`StaticFileJarProvider`].
110///
111/// "Static files" contain immutable chain history data, such as:
112///  - transactions
113///  - headers
114///  - receipts
115///
116/// This provider type is responsible for reading and writing to static files.
117#[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/// Builder for [`StaticFileProvider`] that allows configuration before initialization.
127#[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    /// Creates a new builder with read-write access.
138    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    /// Creates a new builder with read-only access.
149    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    /// Set custom blocks per file for specific segments.
160    ///
161    /// Each static file segment is stored across multiple files, and each of these files contains
162    /// up to the specified number of blocks of data. When the file gets full, a new file is
163    /// created with the new block range.
164    ///
165    /// This setting affects the size of each static file, and can be set per segment.
166    ///
167    /// If it is changed for an existing node, existing static files will not be affected and will
168    /// be finished with the old blocks per file setting, but new static files will use the new
169    /// setting.
170    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    /// Set a custom number of blocks per file for all segments.
181    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    /// Set a custom number of blocks per file for a specific segment.
189    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    /// Enables metrics on the [`StaticFileProvider`].
199    pub const fn with_metrics(mut self) -> Self {
200        self.use_metrics = true;
201        self
202    }
203
204    /// Sets the genesis block number for the [`StaticFileProvider`].
205    ///
206    /// This configures the genesis block number, which is used to determine the starting point
207    /// for block indexing and querying operations.
208    ///
209    /// # Arguments
210    ///
211    /// * `genesis_block_number` - The block number of the genesis block.
212    ///
213    /// # Returns
214    ///
215    /// Returns `Self` to allow method chaining.
216    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    /// Builds the final [`StaticFileProvider`] and initializes the index.
222    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    /// Creates a new [`StaticFileProvider`] with the given [`StaticFileAccess`].
241    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    /// Creates a new [`StaticFileProvider`] with read-only access.
250    ///
251    /// The caller is responsible for calling [`StaticFileProvider::initialize_index`] when
252    /// underlying data changes.
253    pub fn read_only(path: impl AsRef<Path>) -> ProviderResult<Self> {
254        Self::new(path, StaticFileAccess::RO)
255    }
256
257    /// Creates a new [`StaticFileProvider`] with read-write access.
258    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/// [`StaticFileProviderInner`] manages all existing [`StaticFileJarProvider`].
272#[derive(Debug)]
273pub struct StaticFileProviderInner<N> {
274    /// Maintains a map which allows for concurrent access to different `NippyJars`, over different
275    /// segments and ranges.
276    map: DashMap<(BlockNumber, StaticFileSegment), LoadedJar>,
277    /// Indexes per segment.
278    indexes: RwLock<StaticFileMap<StaticFileSegmentIndex>>,
279    /// This is an additional index that tracks the expired height, this will track the highest
280    /// block number that has been expired (missing). The first, non expired block is
281    /// `expired_history_height + 1`.
282    ///
283    /// This is effectively the transaction range that has been expired:
284    /// [`StaticFileProvider::delete_segment_below_block`] and mirrors
285    /// `static_files_min_block[transactions] - blocks_per_file`.
286    ///
287    /// This additional tracker exists for more efficient lookups because the node must be aware of
288    /// the expired height.
289    earliest_history_height: AtomicU64,
290    /// Directory where `static_files` are located
291    path: PathBuf,
292    /// Maintains a writer set of [`StaticFileSegment`].
293    writers: StaticFileWriters<N>,
294    /// Metrics for the static files.
295    metrics: Option<Arc<StaticFileProviderMetrics>>,
296    /// Access rights of the provider.
297    access: StaticFileAccess,
298    /// Number of blocks per file, per segment.
299    blocks_per_file: StaticFileMap<u64>,
300    /// Write lock for when access is [`StaticFileAccess::RW`].
301    _lock_file: Option<StorageLock>,
302    /// Genesis block number, default is 0;
303    genesis_block_number: u64,
304}
305
306impl<N: NodePrimitives> StaticFileProviderInner<N> {
307    /// Creates a new [`StaticFileProviderInner`].
308    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    /// Each static file has a fixed number of blocks. This gives out the range where the requested
341    /// block is positioned.
342    ///
343    /// If the specified block falls into one of the ranges of already initialized static files,
344    /// this function will return that range.
345    ///
346    /// If no matching file exists, this function will derive a new range from the end of the last
347    /// existing file, if any.
348    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            // Find first block range that contains the requested block
359            if let Some((_, range)) = block_index.range(block..).next() {
360                // Found matching range for an existing file using block index
361                return *range;
362            } else if let Some((_, range)) = block_index.last_key_value() {
363                // Didn't find matching range for an existing file, derive a new range from the end
364                // of the last existing file range.
365                //
366                // `block` is always higher than `range.end()` here, because `block_index` holds no
367                // range with a `max_block` greater than or equal to `block`
368                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        // No block index is available, derive a new range using the fixed number of blocks,
375        // starting from the beginning.
376        find_fixed_range(block, blocks_per_file)
377    }
378
379    /// Each static file has a fixed number of blocks. This gives out the range where the requested
380    /// block is positioned.
381    ///
382    /// If the specified block falls into one of the ranges of already initialized static files,
383    /// this function will return that range.
384    ///
385    /// If no matching file exists, this function will derive a new range from the end of the last
386    /// existing file, if any.
387    ///
388    /// This function will block indefinitely if a write lock for
389    /// [`Self::indexes`] is already acquired. In that case, use
390    /// [`Self::find_fixed_range_with_block_index`].
391    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    /// Get genesis block number
404    pub const fn genesis_block_number(&self) -> u64 {
405        self.genesis_block_number
406    }
407}
408
409impl<N: NodePrimitives> StaticFileProvider<N> {
410    /// Reports metrics for the static files.
411    ///
412    /// This uses the in-memory index to get file sizes from mmap handles instead of reading
413    /// filesystem metadata.
414    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    /// Writes headers for all blocks to the static file segment.
441    #[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    /// Writes transactions for all blocks to the static file segment.
454    #[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    /// Writes transaction senders for all blocks to the static file segment.
471    #[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    /// Writes receipts for all blocks to the static file segment.
488    #[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            // skip writing receipts if pruning configuration requires us to.
500            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    /// Writes account changesets for all blocks to the static file segment.
515    #[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    /// Writes storage changesets for all blocks to the static file segment.
536    #[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    /// Writes to a static file segment using the provided closure.
565    ///
566    /// The closure receives a mutable reference to the segment writer. After the closure completes,
567    /// `sync_all()` is called to flush writes to disk.
568    #[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    /// Writes all static file data for multiple blocks in parallel per-segment.
584    ///
585    /// This spawns tasks on the storage thread pool for each segment type and each task calls
586    /// `sync_all()` on its writer when done.
587    #[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        // Propagate tracing context into rayon-spawned threads so that per-segment
609        // write spans appear as children of write_blocks_data in traces.
610        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    /// Gets the [`StaticFileJarProvider`] of the requested segment and start index that can be
694    /// either block or transaction.
695    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    /// Gets the [`StaticFileJarProvider`] of the requested segment and start index that can be
708    /// either block or transaction.
709    ///
710    /// If the segment is not found, returns [`None`].
711    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    /// Gets the [`StaticFileJarProvider`] of the requested segment and block.
733    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    /// Gets the [`StaticFileJarProvider`] of the requested segment and transaction.
748    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    /// Gets the [`StaticFileJarProvider`] of the requested segment and block or transaction.
763    ///
764    /// `fn_range` should make sure the range goes through `find_fixed_range`.
765    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        // If we have a path, then get the block range from its name.
772        // Otherwise, check `self.available_static_files`
773        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        // Return cached `LoadedJar` or insert it for the first time, and then, return it.
792        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    /// Gets the [`StaticFileJarProvider`] of the requested path.
800    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    /// Given a segment and block range it removes the cached provider from the map.
815    ///
816    /// CAUTION: cached provider should be dropped before calling this or IT WILL deadlock.
817    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    /// This handles history expiry by deleting all static files for the given segment below the
826    /// given block.
827    ///
828    /// For example if block is 1M and the blocks per file are 500K this will delete all individual
829    /// files below 1M, so 0-499K and 500K-999K.
830    ///
831    /// This will not delete the file that contains the block itself, because files can only be
832    /// removed entirely.
833    ///
834    /// # Safety
835    ///
836    /// This method will never delete the highest static file for the segment, even if the
837    /// requested block is higher than the highest block in static files. This ensures we always
838    /// maintain at least one static file if any exist.
839    ///
840    /// Returns a list of `SegmentHeader`s from the deleted jars.
841    pub fn delete_segment_below_block(
842        &self,
843        segment: StaticFileSegment,
844        block: BlockNumber,
845    ) -> ProviderResult<Vec<SegmentHeader>> {
846        // Nothing to delete if block is 0.
847        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            // Stop if we've reached the target block or the highest static file
860            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            // now we need to wipe the static file, this will take care of updating the index and
872            // advance the lowest tracked block height for the segment.
873            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    /// Given a segment and block, it deletes the jar and all files from the respective block range.
882    ///
883    /// CAUTION: destructive. Deletes files on disk.
884    ///
885    /// This will re-initialize the index after deletion, so all files are tracked.
886    ///
887    /// Returns the `SegmentHeader` of the deleted jar.
888    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        // Delete the sidecar file for changeset segments before deleting the main jar
912        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        // SAFETY: this is currently necessary to ensure that certain indexes like
922        // `static_files_min_block` have the correct values after pruning.
923        self.initialize_index()?;
924
925        Ok(header)
926    }
927
928    /// Deletes ALL static file jars for the given segment, including the highest one.
929    ///
930    /// CAUTION: destructive. Deletes all files on disk for this segment.
931    ///
932    /// This is used for `PruneMode::Full` where all data should be removed.
933    ///
934    /// Returns a list of `SegmentHeader`s from the deleted jars.
935    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    /// Given a segment and block range it returns a cached
959    /// [`StaticFileJarProvider`]. TODO(joshie): we should check the size and pop N if there's too
960    /// many.
961    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        // Avoid using `entry` directly to avoid a write lock in the common case.
969        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            // The cache may have been populated since the initial miss, including by
979            // `update_index` publishing a newer snapshot while we loaded this jar without a lock.
980            // Preserve that entry instead of overwriting it with our potentially stale snapshot.
981            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    /// Gets a static file segment's block range from the provider inner block
991    /// index.
992    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    /// Gets a static file segment's fixed block range from the provider inner
1010    /// transaction index.
1011    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        // It's more probable that the request comes from a newer tx height, so we iterate
1021        // the static_files in reverse.
1022        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                // request tx is higher than highest static file tx
1027                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    /// Updates the inner transaction and block indexes alongside the internal cached providers in
1042    /// `self.map`.
1043    ///
1044    /// Any entry higher than `segment_max_block` will be deleted from the previous structures.
1045    ///
1046    /// If `segment_max_block` is None it means there's no static file for this segment.
1047    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                        // Update max block
1077                        index.max_block = segment_max_block;
1078
1079                        // Update expected block range index
1080
1081                        // Remove all expected block ranges that are less than the new max block
1082                        index
1083                            .expected_block_ranges_by_max_block
1084                            .retain(|_, block_range| block_range.start() < fixed_range.start());
1085                        // Insert new expected block range
1086                        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                // Update min_block to track the lowest block range of the segment.
1101                // This is initially set by initialize_index() on node startup, but must be updated
1102                // as the file grows to prevent stale values.
1103                //
1104                // Without this update, min_block can remain at genesis (e.g. Some([0..=0]) or None)
1105                // even after syncing to higher blocks (e.g. [0..=100]). A stale
1106                // min_block causes get_lowest_static_file_block() to return the
1107                // wrong end value, which breaks pruning logic that relies on it for
1108                // safety checks.
1109                //
1110                // Example progression:
1111                // 1. Node starts, initialize_index() sets min_block = [0..=0]
1112                // 2. Sync to block 100, this update sets min_block = [0..=100]
1113                // 3. Pruner calls get_lowest_static_file_block() -> returns 100 (correct). Without
1114                //    this update, it would incorrectly return 0 (stale)
1115                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                        // delete_jar WILL ALWAYS re-initialize all indexes, so we are always
1118                        // sure that current_min is always the lowest.
1119                        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                // Updates the tx index by first removing all entries which have a higher
1128                // block_start than our current static file.
1129                if let Some(tx_range) = jar.user_header().tx_range() {
1130                    // Current block range has the same block start as `fixed_range``, but block end
1131                    // might be different if we are still filling this static file.
1132                    if let Some(current_block_range) = jar.user_header().block_range() {
1133                        let tx_end = tx_range.end();
1134
1135                        // Considering that `update_index` is called when we either append/truncate,
1136                        // we are sure that we are handling the latest data
1137                        // points.
1138                        //
1139                        // Here we remove every entry of the index that has a block start higher or
1140                        // equal than our current one. This is important in the case
1141                        // that we prune a lot of rows resulting in a file (and thus
1142                        // a higher block range) deletion.
1143                        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                    // The unwinded file has no more transactions/receipts. However, the highest
1154                    // block is within this files' block range. We only retain
1155                    // entries with block ranges before the current one.
1156                    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                    // If the index is empty, just remove it.
1161                    index.available_block_ranges_by_max_tx.take_if(|index| index.is_empty());
1162                }
1163
1164                // Update the cached provider.
1165                trace!(target: "providers::static_file", ?segment, "Inserting updated jar into cache");
1166                self.map.insert((fixed_range.end(), segment), LoadedJar::new(jar)?);
1167
1168                // Delete any cached provider that no longer has an associated jar.
1169                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    /// Initializes the inner transaction and block index
1183    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            // Update first and last block for each segment
1189            //
1190            // It's safe to call `expect` here, because every segment has at least one header
1191            // associated with it.
1192            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                // Update max expected block -> expected_block_range index
1200                expected_block_ranges_by_max_block
1201                    .insert(header.expected_block_end(), header.expected_block_range());
1202
1203                // Update max tx -> block_range index
1204                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        // If this is a re-initialization, we need to clear this as well
1225        self.map.clear();
1226
1227        // initialize the expired history height to the lowest static file block
1228        if let Some(lowest_range) =
1229            indexes.get(StaticFileSegment::Transactions).and_then(|index| index.min_block_range)
1230        {
1231            // the earliest height is the lowest available block number
1232            self.earliest_history_height
1233                .store(lowest_range.start(), std::sync::atomic::Ordering::Relaxed);
1234        }
1235
1236        Ok(())
1237    }
1238
1239    /// Ensures that any broken invariants which cannot be healed on the spot return a pipeline
1240    /// target to unwind to.
1241    ///
1242    /// Two types of consistency checks are done for:
1243    ///
1244    /// 1) When a static file fails to commit but the underlying data was changed.
1245    /// 2) When a static file was committed, but the required database transaction was not.
1246    ///
1247    /// For 1) it can self-heal if `self.access.is_read_only()` is set to `false`. Otherwise, it
1248    /// will return an error.
1249    /// For 2) the invariants below are checked, and if broken, might require a pipeline unwind
1250    /// to heal.
1251    ///
1252    /// For each static file segment:
1253    /// * the corresponding database table should overlap or have continuity in their keys
1254    ///   ([`TxNumber`] or [`BlockNumber`]).
1255    /// * its highest block should match the stage checkpoint block number if it's equal or higher
1256    ///   than the corresponding database table last entry.
1257    ///
1258    /// Returns a [`Option`] of [`PipelineTarget::Unwind`] if any healing is further required.
1259    ///
1260    /// WARNING: No static file writer should be held before calling this function, otherwise it
1261    /// will deadlock.
1262    #[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        // OVM historical import is broken and does not work with this check. It's importing
1277        // duplicated receipts resulting in having more receipts than the expected transaction
1278        // range.
1279        //
1280        // If we detect an OVM import was done (block #1 <https://optimistic.etherscan.io/block/1>), skip it.
1281        // More on [#11099](https://github.com/paradigmxyz/reth/pull/11099).
1282        if provider.chain_spec().is_optimism() &&
1283            reth_chainspec::Chain::optimism_mainnet() == provider.chain_spec().chain_id()
1284        {
1285            // check whether we have the first OVM block: <https://optimistic.etherscan.io/block/0xbee7192e575af30420cae0c7776304ac196077ee72b048970549e4f08e875453>
1286            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            // Heal file-level inconsistencies and get before/after highest block
1318            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            // Only applies to block-based static files. (Headers)
1323            //
1324            // The updated `highest_block` may have decreased if we healed from a pruning
1325            // interruption.
1326            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            // Only applies to transaction-based static files. (Receipts & Transactions)
1336            //
1337            // Make sure the last transaction matches the last block from its indices, since a heal
1338            // from a pruning interruption might have decreased the number of transactions without
1339            // being able to update the last block of the static file segment.
1340            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                        // If the block body indices can not be found, then it means that static
1351                        // files is ahead of database, and the `ensure_invariants` check will fix
1352                        // it by comparing with stage checkpoints.
1353                        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    /// Heals file-level (`NippyJar`) inconsistencies for eligible static file
1398    /// segments.
1399    ///
1400    /// Call before [`Self::check_consistency`] so files are internally
1401    /// consistent.
1402    ///
1403    /// Uses the same segment-skip logic as [`Self::check_consistency`], but
1404    /// does not compare with database checkpoints or prune against them.
1405    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    /// Returns the static file segments that should be checked/healed for this provider.
1420    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    /// True if the given segment should be checked/healed for this provider.
1432    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                    // Old pruned nodes (including full node) do not store receipts as static
1445                    // files.
1446                    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                    // Gnosis and Chiado's historical import is broken and does not work with
1454                    // this check. They are importing receipts along
1455                    // with importing headers/bodies.
1456                    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    /// Returns `true` if the given prune segment has a checkpoint with
1493    /// [`reth_prune_types::PruneMode::Full`], indicating all data for this segment has been
1494    /// intentionally deleted.
1495    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    /// Checks consistency of the latest static file segment and throws an
1507    /// error if at fault.
1508    ///
1509    /// Read-only.
1510    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    /// Attempts to heal file-level (`NippyJar`) inconsistencies for a single static file segment.
1530    ///
1531    /// Returns the highest block before and after healing, which can be used to detect
1532    /// if healing from a pruning interruption decreased the highest block.
1533    ///
1534    /// File consistency is broken if:
1535    ///
1536    /// * appending data was interrupted before a config commit, then data file will be truncated
1537    ///   according to the config.
1538    ///
1539    /// * pruning data was interrupted before a config commit, then we have deleted data that we are
1540    ///   expected to still have. We need to check the Database and unwind everything accordingly.
1541    ///
1542    /// **Note:** In read-only mode, this will return an error if a consistency issue is detected,
1543    /// since healing requires write access.
1544    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            // Read-only mode: cannot modify files, so just validate consistency and error if
1553            // broken.
1554            debug!(target: "reth::providers::static_file", "Checking segment consistency (read-only)");
1555            self.check_segment_consistency(segment)?;
1556        } else {
1557            // Writable mode: fetching the writer will automatically heal any file-level
1558            // inconsistency by truncating data to match the last committed config.
1559            debug!(target: "reth::providers::static_file", "Fetching latest writer which might heal any potential inconsistency");
1560            self.latest_writer(segment)?;
1561        }
1562
1563        // The updated `highest_block` may have decreased if we healed from a
1564        // pruning interruption.
1565        let highest_block = self.get_highest_static_file_block(segment);
1566
1567        Ok((initial_highest_block, highest_block))
1568    }
1569
1570    /// Ensure invariants for each corresponding table and static file segment.
1571    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    /// Check invariants for each corresponding table and static file segment:
1629    ///
1630    /// * the corresponding database table should overlap or have continuity in their keys
1631    ///   ([`TxNumber`] or [`BlockNumber`]).
1632    /// * its highest block should match the stage checkpoint block number if it's equal or higher
1633    ///   than the corresponding database table last entry.
1634    ///   * If the checkpoint block is higher, then request a pipeline unwind to the static file
1635    ///     block. This is expressed by returning [`Some`] with the requested pipeline unwind
1636    ///     target.
1637    ///   * If the checkpoint block is lower, then heal by removing rows from the static file. In
1638    ///     this case, the rows will be removed and [`None`] will be returned.
1639    ///
1640    /// * If the database tables overlap with static files and have contiguous keys, or the
1641    ///   checkpoint block matches the highest static files block, then [`None`] will be returned.
1642    #[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 there is a gap between the entry found in static file and
1662                // database, then we have most likely lost static file data and need to unwind so we
1663                // can load it again
1664                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        // If static file entry is ahead of the database entries, then ensure the checkpoint block
1691        // number matches.
1692        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 the checkpoint is ahead, then we lost static file data. May be data corruption.
1701        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 the checkpoint is ahead, or matches, then nothing to do.
1712        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        // If the checkpoint is behind, then we failed to do a database commit
1718        // **but committed** to static files on executing a stage, or the
1719        // reverse on unwinding a stage.
1720        //
1721        // All we need to do is to prune the extra static file rows.
1722        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                // TODO(joshie): is_block_meta
1735                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    /// Returns the highest block accounted for in this segment, either through data
1876    /// present in static files or through data intentionally removed by pruning.
1877    ///
1878    /// Data below a segment's prune checkpoint has been intentionally deleted, so its
1879    /// absence from static files is not an inconsistency. Without this, a pruned segment
1880    /// whose stage checkpoint is ahead of its (empty) static files is treated as data
1881    /// corruption and triggers an unwind to block 0, which aborts the node on startup.
1882    /// See <https://github.com/paradigmxyz/reth/issues/23463>.
1883    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    /// Returns the prune segment that governs data availability for a static file segment,
1904    /// or `None` if the segment is never pruned.
1905    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    /// Returns the earliest available block number that has not been expired and is still
1916    /// available.
1917    ///
1918    /// This means that the highest expired block (or expired block height) is
1919    /// `earliest_history_height.saturating_sub(1)`.
1920    ///
1921    /// Returns `0` if no history has been expired.
1922    pub fn earliest_history_height(&self) -> BlockNumber {
1923        self.earliest_history_height.load(std::sync::atomic::Ordering::Relaxed)
1924    }
1925
1926    /// Gets the lowest static file's block range if it exists for a static file segment.
1927    ///
1928    /// If there is nothing on disk for the given segment, this will return [`None`].
1929    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    /// Gets the lowest static file's block range start if it exists for a static file segment.
1934    ///
1935    /// For example if the lowest static file has blocks 0-499, this will return 0.
1936    ///
1937    /// If there is nothing on disk for the given segment, this will return [`None`].
1938    pub fn get_lowest_range_start(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
1939        self.get_lowest_range(segment).map(|range| range.start())
1940    }
1941
1942    /// Gets the lowest static file's block range end if it exists for a static file segment.
1943    ///
1944    /// For example if the static file has blocks 0-499, this will return 499.
1945    ///
1946    /// If there is nothing on disk for the given segment, this will return [`None`].
1947    pub fn get_lowest_range_end(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
1948        self.get_lowest_range(segment).map(|range| range.end())
1949    }
1950
1951    /// Gets the highest static file's block height if it exists for a static file segment.
1952    ///
1953    /// If there is nothing on disk for the given segment, this will return [`None`].
1954    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    /// Converts a range to a bounded `RangeInclusive` capped to the highest static file block.
1959    ///
1960    /// This is necessary because static file iteration beyond the tip would loop forever:
1961    /// blocks beyond the static file tip return `Ok(empty)` which is indistinguishable from
1962    /// blocks with no changes. We cap the end to the highest available block regardless of
1963    /// whether the input was unbounded or an explicit large value like `BlockNumber::MAX`.
1964    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    /// Gets the highest static file transaction.
1986    ///
1987    /// If there is nothing on disk for the given segment, this will return [`None`].
1988    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    /// Gets the highest static file block for all segments.
1997    pub fn get_highest_static_files(&self) -> HighestStaticFiles {
1998        HighestStaticFiles {
1999            receipts: self.get_highest_static_file_block(StaticFileSegment::Receipts),
2000        }
2001    }
2002
2003    /// Iterates through segment `static_files` in reverse order, executing a function until it
2004    /// returns some object. Useful for finding objects by [`TxHash`] or [`BlockHash`].
2005    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            // Iterate through all ranges in reverse order (highest to lowest)
2014            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    /// Fetches data within a specified range across multiple static files.
2025    ///
2026    /// This function iteratively retrieves data using `get_fn` for each item in the given range.
2027    /// It continues fetching until the end of the range is reached or the provided `predicate`
2028    /// returns false.
2029    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        /// Resolves to the provider for the given block or transaction number.
2043        ///
2044        /// If the static file is missing, the `result` is returned.
2045        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        // advances number in range
2062        'outer: for number in range {
2063            // The `retrying` flag ensures a single retry attempt per `number`. If `get_fn` fails to
2064            // access data in two different static files, it halts further attempts by returning
2065            // an error, effectively preventing infinite retry loops.
2066            let mut retrying = false;
2067
2068            // advances static files if `get_fn` returns None
2069            '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                        // There is a very small chance of hitting a deadlock if two consecutive
2083                        // static files share the same bucket in the
2084                        // internal dashmap and we don't drop the current provider
2085                        // before requesting the next one.
2086                        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    /// Fetches data within a specified range across multiple static files.
2102    ///
2103    /// Returns an iterator over the data. Yields [`None`] if the data for the specified number is
2104    /// not found.
2105    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                    // There is a very small chance of hitting a deadlock if two consecutive
2125                    // static files share the same bucket in the internal dashmap and we don't drop
2126                    // the current provider before requesting the next one.
2127                    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    /// Returns directory where `static_files` are located.
2140    pub fn directory(&self) -> &Path {
2141        &self.path
2142    }
2143
2144    /// Retrieves data from the database or static file, wherever it's available.
2145    ///
2146    /// # Arguments
2147    /// * `segment` - The segment of the static file to check against.
2148    /// * `index_key` - Requested index key, usually a block or transaction number.
2149    /// * `fetch_from_static_file` - A closure that defines how to fetch the data from the static
2150    ///   file provider.
2151    /// * `fetch_from_database` - A closure that defines how to fetch the data from the database
2152    ///   when the static file doesn't contain the required data or is not available.
2153    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        // If there is, check the maximum block or transaction number of the segment.
2165        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    /// Gets data within a specified range, potentially spanning different `static_files` and
2180    /// database.
2181    ///
2182    /// # Arguments
2183    /// * `segment` - The segment of the static file to query.
2184    /// * `block_or_tx_range` - The range of data to fetch.
2185    /// * `fetch_from_static_file` - A function to fetch data from the `static_file`.
2186    /// * `fetch_from_database` - A function to fetch data from the database.
2187    /// * `predicate` - A function used to evaluate each item in the fetched data. Fetching is
2188    ///   terminated when this function returns false, thereby filtering the data based on the
2189    ///   provided condition.
2190    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 there is, check the maximum block or transaction number of the segment.
2206        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    /// Returns static files directory
2229    #[cfg(any(test, feature = "test-utils"))]
2230    pub fn path(&self) -> &Path {
2231        &self.path
2232    }
2233
2234    /// Returns transaction index
2235    #[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    /// Returns expected block index
2245    #[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 static file block range.
2258    ///
2259    /// This index is initialized on launch to keep track of the lowest, non-expired static file
2260    /// per segment and gets updated on [`StaticFileProvider::update_index`].
2261    ///
2262    /// This tracks the lowest static file per segment together with the block range in that
2263    /// file. E.g. static file is batched in 500k block intervals then the lowest static file
2264    /// is [0..499K], and the block range is start = 0, end = 499K.
2265    ///
2266    /// This index is mainly used for history expiry, which targets transactions, e.g. pre-merge
2267    /// history expiry would lead to removing all static files below the merge height.
2268    min_block_range: Option<SegmentRangeInclusive>,
2269    /// Max static file block.
2270    max_block: u64,
2271    /// Expected static file block ranges indexed by max expected blocks.
2272    ///
2273    /// For example, a static file for expected block range `0..=499_000` may have only block range
2274    /// `0..=1000` contained in it, as it's not fully filled yet. This index maps the max expected
2275    /// block to the expected range, i.e. block `499_000` to block range `0..=499_000`.
2276    expected_block_ranges_by_max_block: SegmentRanges,
2277    /// Available on disk static file block ranges indexed by max transactions.
2278    ///
2279    /// For example, a static file for block range `0..=499_000` may only have block range
2280    /// `0..=1000` and transaction range `0..=2000` contained in it. This index maps the max
2281    /// available transaction to the available block range, i.e. transaction `2000` to block range
2282    /// `0..=1000`.
2283    available_block_ranges_by_max_tx: Option<SegmentRanges>,
2284}
2285
2286/// Helper trait to manage different [`StaticFileProviderRW`] of an `Arc<StaticFileProvider`
2287pub trait StaticFileWriter {
2288    /// The primitives type used by the static file provider.
2289    type Primitives: Send + Sync + 'static;
2290
2291    /// Returns a mutable reference to a [`StaticFileProviderRW`] of a [`StaticFileSegment`].
2292    fn get_writer(
2293        &self,
2294        block: BlockNumber,
2295        segment: StaticFileSegment,
2296    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2297
2298    /// Returns a mutable reference to a [`StaticFileProviderRW`] of the latest
2299    /// [`StaticFileSegment`].
2300    fn latest_writer(
2301        &self,
2302        segment: StaticFileSegment,
2303    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2304
2305    /// Commits all changes of all [`StaticFileProviderRW`] of all [`StaticFileSegment`].
2306    fn commit(&self) -> ProviderResult<()>;
2307
2308    /// Returns `true` if the static file provider has unwind queued.
2309    fn has_unwind_queued(&self) -> bool;
2310
2311    /// Finalizes all static file writers by committing their configuration to disk.
2312    ///
2313    /// Returns an error if prune is queued (use [`Self::commit`] instead).
2314    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                // This is not expected but means we are out of the range / file somehow, and can't
2427                // continue
2428                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    /// Creates an iterator for walking through account changesets in the specified block range.
2566    ///
2567    /// This returns a lazy iterator that fetches changesets block by block to avoid loading
2568    /// everything into memory at once.
2569    ///
2570    /// Accepts any range type that implements `RangeBounds<BlockNumber>`, including:
2571    /// - `Range<BlockNumber>` (e.g., `0..100`)
2572    /// - `RangeInclusive<BlockNumber>` (e.g., `0..=99`)
2573    /// - `RangeFrom<BlockNumber>` (e.g., `0..`) - iterates until exhausted
2574    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    /// Creates an iterator for walking through storage changesets in the specified block range.
2582    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        // Transactions are different size, so chunks will not all take the same processing time. If
2752        // chunks are too big, there will be idle threads waiting for work. Choosing an
2753        // arbitrary smaller value to make sure it doesn't happen.
2754        let chunk_size = 100;
2755
2756        // iterator over the chunks
2757        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            // Spawn the task onto the global rayon pool
2770            // This task will send the cached transaction hash through the channel.
2771            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        // Iterate over channels and append the tx hashes unsorted
2790        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        // Required data not present in static_files
2862        Err(ProviderError::UnsupportedProvider)
2863    }
2864
2865    fn transactions_by_block(
2866        &self,
2867        _block_id: BlockHashOrNumber,
2868    ) -> ProviderResult<Option<Vec<Self::Transaction>>> {
2869        // Required data not present in static_files
2870        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        // Required data not present in static_files
2878        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        // Required data not present in static_files
2921        Err(ProviderError::UnsupportedProvider)
2922    }
2923
2924    fn best_block_number(&self) -> ProviderResult<BlockNumber> {
2925        // Required data not present in static_files
2926        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        // Required data not present in static_files
2935        Err(ProviderError::UnsupportedProvider)
2936    }
2937}
2938
2939/* Cannot be successfully implemented but must exist for trait requirements */
2940
2941impl<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        // Required data not present in static_files
2952        Err(ProviderError::UnsupportedProvider)
2953    }
2954
2955    fn block(&self, _id: BlockHashOrNumber) -> ProviderResult<Option<Self::Block>> {
2956        // Required data not present in static_files
2957        Err(ProviderError::UnsupportedProvider)
2958    }
2959
2960    fn pending_block(&self) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
2961        // Required data not present in static_files
2962        Err(ProviderError::UnsupportedProvider)
2963    }
2964
2965    fn pending_block_and_receipts(
2966        &self,
2967    ) -> ProviderResult<Option<(RecoveredBlock<Self::Block>, Vec<Self::Receipt>)>> {
2968        // Required data not present in static_files
2969        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        // Required data not present in static_files
2978        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        // Required data not present in static_files
2987        Err(ProviderError::UnsupportedProvider)
2988    }
2989
2990    fn block_range(&self, _range: RangeInclusive<BlockNumber>) -> ProviderResult<Vec<Self::Block>> {
2991        // Required data not present in static_files
2992        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/// Returns the tx hash for the given transaction and its id.
3056#[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        // Test with None - should use default behavior
3084        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        // Test with empty index - should fall back to default behavior
3094        assert_eq!(
3095            sf_rw.find_fixed_range_with_block_index(segment, Some(&BTreeMap::new()), 150),
3096            SegmentRangeInclusive::new(100, 199)
3097        );
3098
3099        // Create block index with existing ranges
3100        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        // Test blocks within existing ranges - should return the matching range
3107        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        // Test blocks beyond existing ranges - should derive new ranges from the last range
3133        // Block 300 is exactly one segment after the last range
3134        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        // Block 500 skips one segment (300-399)
3144        assert_eq!(
3145            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 500),
3146            SegmentRangeInclusive::new(500, 599)
3147        );
3148
3149        // Block 1000 skips many segments
3150        assert_eq!(
3151            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 1000),
3152            SegmentRangeInclusive::new(1000, 1099)
3153        );
3154
3155        // Test with block index having different sizes than blocks_per_file setting
3156        // This simulates the scenario where blocks_per_file was changed between runs
3157        let mixed_size_index = BTreeMap::from_iter([
3158            (49, SegmentRangeInclusive::new(0, 49)),     // 50 blocks
3159            (149, SegmentRangeInclusive::new(50, 149)),  // 100 blocks
3160            (349, SegmentRangeInclusive::new(150, 349)), // 200 blocks
3161        ]);
3162
3163        // Blocks within existing ranges should return those ranges regardless of size
3164        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        // Block after the last range should derive using current blocks_per_file (100)
3178        // from the end of the last range (349)
3179        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}