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_execution_types::RecoveredBlockAndExecutionOutput;
37use reth_nippy_jar::{NippyJar, NippyJarChecker};
38use reth_node_types::NodePrimitives;
39use reth_primitives_traits::{
40    dashmap::DashMap, AlloyBlockHeader as _, BlockBody as _, RecoveredBlock, SealedHeader,
41    SignedTransaction, StorageEntry,
42};
43use reth_prune_types::PruneSegment;
44use reth_stages_types::PipelineTarget;
45use reth_static_file_types::{
46    find_fixed_range, HighestStaticFiles, SegmentHeader, SegmentRangeInclusive, StaticFileMap,
47    StaticFileSegment, DEFAULT_BLOCKS_PER_STATIC_FILE,
48};
49use reth_storage_api::{
50    BlockBodyIndicesProvider, ChangeSetReader, DBProvider, PruneCheckpointReader,
51    StorageChangeSetReader, StorageSettingsCache,
52};
53use reth_storage_errors::provider::{ProviderError, ProviderResult, StaticFileWriterError};
54use std::{
55    collections::BTreeMap,
56    fmt::Debug,
57    ops::{Bound, Deref, Range, RangeBounds, RangeInclusive},
58    path::{Path, PathBuf},
59    sync::{
60        atomic::{AtomicU64, Ordering},
61        mpsc, Arc,
62    },
63};
64use tracing::{debug, info, info_span, instrument, trace, warn};
65
66/// Alias type for a map that can be queried for block or transaction ranges. It uses `u64` to
67/// represent either a block or a transaction number end of a static file range.
68type SegmentRanges = BTreeMap<u64, SegmentRangeInclusive>;
69
70/// Access mode on a static file provider. RO/RW.
71#[derive(Debug, Default, PartialEq, Eq)]
72pub enum StaticFileAccess {
73    /// Read-only access.
74    #[default]
75    RO,
76    /// Read-write access.
77    RW,
78}
79
80impl StaticFileAccess {
81    /// Returns `true` if read-only access.
82    pub const fn is_read_only(&self) -> bool {
83        matches!(self, Self::RO)
84    }
85
86    /// Returns `true` if read-write access.
87    pub const fn is_read_write(&self) -> bool {
88        matches!(self, Self::RW)
89    }
90}
91
92/// Context for static file block writes.
93///
94/// Contains target segments and pruning configuration.
95#[derive(Debug, Clone, Copy, Default)]
96pub struct StaticFileWriteCtx {
97    /// Whether transaction senders should be written to static files.
98    pub write_senders: bool,
99    /// Whether receipts should be written to static files.
100    pub write_receipts: bool,
101    /// Whether account changesets should be written to static files.
102    pub write_account_changesets: bool,
103    /// Whether storage changesets should be written to static files.
104    pub write_storage_changesets: bool,
105    /// The current chain tip block number (for pruning).
106    pub tip: BlockNumber,
107    /// The prune mode for receipts, if any.
108    pub receipts_prune_mode: Option<reth_prune_types::PruneMode>,
109    /// Whether receipts are prunable (based on storage settings and prune distance).
110    pub receipts_prunable: bool,
111}
112
113/// [`StaticFileProvider`] manages all existing [`StaticFileJarProvider`].
114///
115/// "Static files" contain immutable chain history data, such as:
116///  - transactions
117///  - headers
118///  - receipts
119///
120/// This provider type is responsible for reading and writing to static files.
121#[derive(Debug)]
122pub struct StaticFileProvider<N>(pub(crate) Arc<StaticFileProviderInner<N>>);
123
124impl<N> Clone for StaticFileProvider<N> {
125    fn clone(&self) -> Self {
126        Self(self.0.clone())
127    }
128}
129
130/// Builder for [`StaticFileProvider`] that allows configuration before initialization.
131#[derive(Debug)]
132pub struct StaticFileProviderBuilder<P> {
133    access: StaticFileAccess,
134    use_metrics: bool,
135    blocks_per_file: StaticFileMap<u64>,
136    path: P,
137    genesis_block_number: u64,
138}
139
140impl<P: AsRef<Path>> StaticFileProviderBuilder<P> {
141    /// Creates a new builder with read-write access.
142    pub fn read_write(path: P) -> Self {
143        Self {
144            path,
145            access: StaticFileAccess::RW,
146            blocks_per_file: Default::default(),
147            use_metrics: false,
148            genesis_block_number: 0,
149        }
150    }
151
152    /// Creates a new builder with read-only access.
153    pub fn read_only(path: P) -> Self {
154        Self {
155            path,
156            access: StaticFileAccess::RO,
157            blocks_per_file: Default::default(),
158            use_metrics: false,
159            genesis_block_number: 0,
160        }
161    }
162
163    /// Set custom blocks per file for specific segments.
164    ///
165    /// Each static file segment is stored across multiple files, and each of these files contains
166    /// up to the specified number of blocks of data. When the file gets full, a new file is
167    /// created with the new block range.
168    ///
169    /// This setting affects the size of each static file, and can be set per segment.
170    ///
171    /// If it is changed for an existing node, existing static files will not be affected and will
172    /// be finished with the old blocks per file setting, but new static files will use the new
173    /// setting.
174    pub fn with_blocks_per_file_for_segments(
175        mut self,
176        segments: &<StaticFileMap<u64> as Deref>::Target,
177    ) -> Self {
178        for (segment, &blocks_per_file) in segments {
179            self.blocks_per_file.insert(segment, blocks_per_file);
180        }
181        self
182    }
183
184    /// Set a custom number of blocks per file for all segments.
185    pub fn with_blocks_per_file(mut self, blocks_per_file: u64) -> Self {
186        for segment in StaticFileSegment::iter() {
187            self.blocks_per_file.insert(segment, blocks_per_file);
188        }
189        self
190    }
191
192    /// Set a custom number of blocks per file for a specific segment.
193    pub fn with_blocks_per_file_for_segment(
194        mut self,
195        segment: StaticFileSegment,
196        blocks_per_file: u64,
197    ) -> Self {
198        self.blocks_per_file.insert(segment, blocks_per_file);
199        self
200    }
201
202    /// Enables metrics on the [`StaticFileProvider`].
203    pub const fn with_metrics(mut self) -> Self {
204        self.use_metrics = true;
205        self
206    }
207
208    /// Sets the genesis block number for the [`StaticFileProvider`].
209    ///
210    /// This configures the genesis block number, which is used to determine the starting point
211    /// for block indexing and querying operations.
212    ///
213    /// # Arguments
214    ///
215    /// * `genesis_block_number` - The block number of the genesis block.
216    ///
217    /// # Returns
218    ///
219    /// Returns `Self` to allow method chaining.
220    pub const fn with_genesis_block_number(mut self, genesis_block_number: u64) -> Self {
221        self.genesis_block_number = genesis_block_number;
222        self
223    }
224
225    /// Builds the final [`StaticFileProvider`] and initializes the index.
226    pub fn build<N: NodePrimitives>(self) -> ProviderResult<StaticFileProvider<N>> {
227        let mut provider = StaticFileProviderInner::new(self.path, self.access)?;
228        if self.use_metrics {
229            provider.metrics = Some(Arc::new(StaticFileProviderMetrics::default()));
230        }
231
232        for (segment, blocks_per_file) in *self.blocks_per_file {
233            provider.blocks_per_file.insert(segment, blocks_per_file);
234        }
235        provider.genesis_block_number = self.genesis_block_number;
236
237        let provider = StaticFileProvider(Arc::new(provider));
238        provider.initialize_index()?;
239        Ok(provider)
240    }
241}
242
243impl<N: NodePrimitives> StaticFileProvider<N> {
244    /// Creates a new [`StaticFileProvider`] with the given [`StaticFileAccess`].
245    fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
246        let provider = Self(Arc::new(StaticFileProviderInner::new(path, access)?));
247        provider.initialize_index()?;
248        Ok(provider)
249    }
250}
251
252impl<N: NodePrimitives> StaticFileProvider<N> {
253    /// Creates a new [`StaticFileProvider`] with read-only access.
254    ///
255    /// The caller is responsible for calling [`StaticFileProvider::initialize_index`] when
256    /// underlying data changes.
257    pub fn read_only(path: impl AsRef<Path>) -> ProviderResult<Self> {
258        Self::new(path, StaticFileAccess::RO)
259    }
260
261    /// Creates a new [`StaticFileProvider`] with read-write access.
262    pub fn read_write(path: impl AsRef<Path>) -> ProviderResult<Self> {
263        Self::new(path, StaticFileAccess::RW)
264    }
265}
266
267impl<N: NodePrimitives> Deref for StaticFileProvider<N> {
268    type Target = StaticFileProviderInner<N>;
269
270    fn deref(&self) -> &Self::Target {
271        &self.0
272    }
273}
274
275#[cfg(test)]
276type BeforeCacheInsert = Box<dyn FnOnce(StaticFileSegment) + Send + 'static>;
277
278#[cfg(test)]
279#[derive(Default)]
280struct BeforeCacheInsertHook(parking_lot::Mutex<Option<BeforeCacheInsert>>);
281
282#[cfg(test)]
283impl Debug for BeforeCacheInsertHook {
284    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
285        f.debug_struct("BeforeCacheInsertHook").finish_non_exhaustive()
286    }
287}
288
289#[cfg(test)]
290impl BeforeCacheInsertHook {
291    fn set(&self, hook: impl FnOnce(StaticFileSegment) + Send + 'static) {
292        let previous = self.0.lock().replace(Box::new(hook));
293        assert!(previous.is_none(), "cache fill hook already installed");
294    }
295
296    fn fire(&self, segment: StaticFileSegment) {
297        if let Some(hook) = self.0.lock().take() {
298            hook(segment);
299        }
300    }
301}
302
303/// [`StaticFileProviderInner`] manages all existing [`StaticFileJarProvider`].
304#[derive(Debug)]
305pub struct StaticFileProviderInner<N> {
306    /// Maintains a map which allows for concurrent access to different `NippyJars`, over different
307    /// segments and ranges.
308    map: DashMap<(BlockNumber, StaticFileSegment), LoadedJar>,
309    /// Incremented before clearing the jar cache so in-flight loads cannot repopulate it with a
310    /// snapshot from before the invalidation.
311    cache_generation: AtomicU64,
312    /// Indexes per segment.
313    indexes: RwLock<StaticFileMap<StaticFileSegmentIndex>>,
314    /// This is an additional index that tracks the expired height, this will track the highest
315    /// block number that has been expired (missing). The first, non expired block is
316    /// `expired_history_height + 1`.
317    ///
318    /// This is effectively the transaction range that has been expired:
319    /// [`StaticFileProvider::delete_segment_below_block`] and mirrors
320    /// `static_files_min_block[transactions] - blocks_per_file`.
321    ///
322    /// This additional tracker exists for more efficient lookups because the node must be aware of
323    /// the expired height.
324    earliest_history_height: AtomicU64,
325    /// Directory where `static_files` are located
326    path: PathBuf,
327    /// Maintains a writer set of [`StaticFileSegment`].
328    writers: StaticFileWriters<N>,
329    /// Metrics for the static files.
330    metrics: Option<Arc<StaticFileProviderMetrics>>,
331    /// Access rights of the provider.
332    access: StaticFileAccess,
333    /// Number of blocks per file, per segment.
334    blocks_per_file: StaticFileMap<u64>,
335    /// Write lock for when access is [`StaticFileAccess::RW`].
336    _lock_file: Option<StorageLock>,
337    /// Genesis block number, default is 0;
338    genesis_block_number: u64,
339    #[cfg(test)]
340    before_cache_insert: BeforeCacheInsertHook,
341    #[cfg(test)]
342    after_cache_validation: BeforeCacheInsertHook,
343}
344
345impl<N: NodePrimitives> StaticFileProviderInner<N> {
346    /// Creates a new [`StaticFileProviderInner`].
347    fn new(path: impl AsRef<Path>, access: StaticFileAccess) -> ProviderResult<Self> {
348        let _lock_file = if access.is_read_write() {
349            StorageLock::try_acquire(path.as_ref()).map_err(ProviderError::other)?.into()
350        } else {
351            None
352        };
353
354        let mut blocks_per_file = StaticFileMap::default();
355        for segment in StaticFileSegment::iter() {
356            blocks_per_file.insert(segment, DEFAULT_BLOCKS_PER_STATIC_FILE);
357        }
358
359        let provider = Self {
360            map: Default::default(),
361            cache_generation: Default::default(),
362            indexes: Default::default(),
363            writers: Default::default(),
364            earliest_history_height: Default::default(),
365            path: path.as_ref().to_path_buf(),
366            metrics: None,
367            access,
368            blocks_per_file,
369            _lock_file,
370            genesis_block_number: 0,
371            #[cfg(test)]
372            before_cache_insert: Default::default(),
373            #[cfg(test)]
374            after_cache_validation: Default::default(),
375        };
376
377        Ok(provider)
378    }
379
380    pub const fn is_read_only(&self) -> bool {
381        self.access.is_read_only()
382    }
383
384    /// Each static file has a fixed number of blocks. This gives out the range where the requested
385    /// block is positioned.
386    ///
387    /// If the specified block falls into one of the ranges of already initialized static files,
388    /// this function will return that range.
389    ///
390    /// If no matching file exists, this function will derive a new range from the end of the last
391    /// existing file, if any.
392    pub fn find_fixed_range_with_block_index(
393        &self,
394        segment: StaticFileSegment,
395        block_index: Option<&SegmentRanges>,
396        block: BlockNumber,
397    ) -> SegmentRangeInclusive {
398        let blocks_per_file =
399            self.blocks_per_file.get(segment).copied().unwrap_or(DEFAULT_BLOCKS_PER_STATIC_FILE);
400
401        if let Some(block_index) = block_index {
402            // Find first block range that contains the requested block
403            if let Some((_, range)) = block_index.range(block..).next() {
404                // Found matching range for an existing file using block index
405                return *range;
406            } else if let Some((_, range)) = block_index.last_key_value() {
407                // Didn't find matching range for an existing file, derive a new range from the end
408                // of the last existing file range.
409                //
410                // `block` is always higher than `range.end()` here, because `block_index` holds no
411                // range with a `max_block` greater than or equal to `block`
412                let blocks_after_last_range = block - range.end();
413                let segments_to_skip = (blocks_after_last_range - 1) / blocks_per_file;
414                let start = range.end() + 1 + segments_to_skip * blocks_per_file;
415                return SegmentRangeInclusive::new(start, start + blocks_per_file - 1);
416            }
417        }
418        // No block index is available, derive a new range using the fixed number of blocks,
419        // starting from the beginning.
420        find_fixed_range(block, blocks_per_file)
421    }
422
423    /// Each static file has a fixed number of blocks. This gives out the range where the requested
424    /// block is positioned.
425    ///
426    /// If the specified block falls into one of the ranges of already initialized static files,
427    /// this function will return that range.
428    ///
429    /// If no matching file exists, this function will derive a new range from the end of the last
430    /// existing file, if any.
431    ///
432    /// This function will block indefinitely if a write lock for
433    /// [`Self::indexes`] is already acquired. In that case, use
434    /// [`Self::find_fixed_range_with_block_index`].
435    pub fn find_fixed_range(
436        &self,
437        segment: StaticFileSegment,
438        block: BlockNumber,
439    ) -> SegmentRangeInclusive {
440        self.find_fixed_range_with_block_index(
441            segment,
442            self.indexes.read().get(segment).map(|index| &index.expected_block_ranges_by_max_block),
443            block,
444        )
445    }
446
447    /// Get genesis block number
448    pub const fn genesis_block_number(&self) -> u64 {
449        self.genesis_block_number
450    }
451}
452
453impl<N: NodePrimitives> StaticFileProvider<N> {
454    /// Reports metrics for the static files.
455    ///
456    /// This uses the in-memory index to get file sizes from mmap handles instead of reading
457    /// filesystem metadata.
458    pub fn report_metrics(&self) -> ProviderResult<()> {
459        let Some(metrics) = &self.metrics else { return Ok(()) };
460
461        let static_files = iter_static_files(&self.path).map_err(ProviderError::other)?;
462        for (segment, headers) in &*static_files {
463            let mut entries = 0;
464            let mut size = 0;
465
466            for (block_range, _) in headers {
467                let fixed_block_range = self.find_fixed_range(segment, block_range.start());
468                let jar_provider = self
469                    .get_segment_provider_for_range(segment, || Some(fixed_block_range), None)?
470                    .ok_or_else(|| {
471                        ProviderError::MissingStaticFileBlock(segment, block_range.start())
472                    })?;
473
474                entries += jar_provider.rows();
475                size += jar_provider.size() as u64;
476            }
477
478            metrics.record_segment(segment, size, headers.len(), entries);
479        }
480
481        Ok(())
482    }
483
484    /// Writes headers for all blocks to the static file segment.
485    #[instrument(level = "debug", target = "providers::static_file", skip_all)]
486    fn write_headers(
487        w: &mut StaticFileProviderRWRefMut<'_, N>,
488        blocks: &[ExecutedBlock<N>],
489    ) -> ProviderResult<()> {
490        for block in blocks {
491            let b = block.recovered_block();
492            w.append_header(b.header(), &b.hash())?;
493        }
494        Ok(())
495    }
496
497    /// Writes transactions for all blocks to the static file segment.
498    #[instrument(level = "debug", target = "providers::static_file", skip_all)]
499    fn write_transactions(
500        w: &mut StaticFileProviderRWRefMut<'_, N>,
501        blocks: &[ExecutedBlock<N>],
502        tx_nums: &[TxNumber],
503    ) -> ProviderResult<()> {
504        for (block, &first_tx) in blocks.iter().zip(tx_nums) {
505            let b = block.recovered_block();
506            w.increment_block(b.number())?;
507            for (i, tx) in b.body().transactions().iter().enumerate() {
508                w.append_transaction(first_tx + i as u64, tx)?;
509            }
510        }
511        Ok(())
512    }
513
514    /// Writes transaction senders for all blocks to the static file segment.
515    #[instrument(level = "debug", target = "providers::static_file", skip_all)]
516    fn write_transaction_senders(
517        w: &mut StaticFileProviderRWRefMut<'_, N>,
518        blocks: &[ExecutedBlock<N>],
519        tx_nums: &[TxNumber],
520    ) -> ProviderResult<()> {
521        for (block, &first_tx) in blocks.iter().zip(tx_nums) {
522            let b = block.recovered_block();
523            w.increment_block(b.number())?;
524            for (i, sender) in b.senders_iter().enumerate() {
525                w.append_transaction_sender(first_tx + i as u64, sender)?;
526            }
527        }
528        Ok(())
529    }
530
531    /// Writes receipts for all blocks to the static file segment.
532    #[instrument(level = "debug", target = "providers::static_file", skip_all)]
533    fn write_receipts(
534        w: &mut StaticFileProviderRWRefMut<'_, N>,
535        blocks: &[ExecutedBlock<N>],
536        tx_nums: &[TxNumber],
537        ctx: &StaticFileWriteCtx,
538    ) -> ProviderResult<()> {
539        for (block, &first_tx) in blocks.iter().zip(tx_nums) {
540            let block_number = block.recovered_block().number();
541            w.increment_block(block_number)?;
542
543            // skip writing receipts if pruning configuration requires us to.
544            if ctx.receipts_prunable &&
545                ctx.receipts_prune_mode
546                    .is_some_and(|mode| mode.should_prune(block_number, ctx.tip))
547            {
548                continue
549            }
550
551            for (i, receipt) in block.execution_outcome().receipts.iter().enumerate() {
552                w.append_receipt(first_tx + i as u64, receipt)?;
553            }
554        }
555        Ok(())
556    }
557
558    /// Writes account changesets for all blocks to the static file segment.
559    #[instrument(level = "debug", target = "providers::static_file", skip_all)]
560    fn write_account_changesets(
561        w: &mut StaticFileProviderRWRefMut<'_, N>,
562        blocks: &[ExecutedBlock<N>],
563    ) -> ProviderResult<()> {
564        for block in blocks {
565            let block_number = block.recovered_block().number();
566            let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
567
568            let changeset: Vec<_> = reverts
569                .accounts
570                .into_iter()
571                .flatten()
572                .map(|(address, info)| AccountBeforeTx { address, info: info.map(Into::into) })
573                .collect();
574            w.append_account_changeset(changeset, block_number)?;
575        }
576        Ok(())
577    }
578
579    /// Writes storage changesets for all blocks to the static file segment.
580    #[instrument(level = "debug", target = "providers::db", skip_all)]
581    fn write_storage_changesets(
582        w: &mut StaticFileProviderRWRefMut<'_, N>,
583        blocks: &[ExecutedBlock<N>],
584    ) -> ProviderResult<()> {
585        for block in blocks {
586            let block_number = block.recovered_block().number();
587            let reverts = block.execution_outcome().state.reverts.to_plain_state_reverts();
588
589            let changeset: Vec<_> = reverts
590                .storage
591                .into_iter()
592                .flatten()
593                .flat_map(|revert| {
594                    revert.storage_revert.into_iter().map(move |(key, revert_to_slot)| {
595                        StorageBeforeTx {
596                            address: revert.address,
597                            key: key.into(),
598                            value: revert_to_slot.to_previous_value(),
599                        }
600                    })
601                })
602                .collect();
603            w.append_storage_changeset(changeset, block_number)?;
604        }
605        Ok(())
606    }
607
608    /// Writes to a static file segment using the provided closure.
609    ///
610    /// The closure receives a mutable reference to the segment writer. After the closure completes,
611    /// `sync_all()` is called to flush writes to disk.
612    #[instrument(level = "debug", target = "providers::static_file", skip_all, fields(?segment))]
613    fn write_segment<F>(
614        &self,
615        segment: StaticFileSegment,
616        first_block_number: BlockNumber,
617        f: F,
618    ) -> ProviderResult<()>
619    where
620        F: FnOnce(&mut StaticFileProviderRWRefMut<'_, N>) -> ProviderResult<()>,
621    {
622        let mut w = self.get_writer(first_block_number, segment)?;
623        f(&mut w)?;
624        w.sync_all()
625    }
626
627    /// Writes all static file data for multiple blocks in parallel per-segment.
628    ///
629    /// This spawns tasks on the storage thread pool for each segment type and each task calls
630    /// `sync_all()` on its writer when done.
631    #[instrument(level = "debug", target = "providers::static_file", skip_all)]
632    pub fn write_blocks_data(
633        &self,
634        blocks: &[ExecutedBlock<N>],
635        tx_nums: &[TxNumber],
636        ctx: StaticFileWriteCtx,
637        runtime: &reth_tasks::Runtime,
638    ) -> ProviderResult<()> {
639        if blocks.is_empty() {
640            return Ok(());
641        }
642
643        let first_block_number = blocks[0].recovered_block().number();
644
645        let mut r_headers = None;
646        let mut r_txs = None;
647        let mut r_senders = None;
648        let mut r_receipts = None;
649        let mut r_account_changesets = None;
650        let mut r_storage_changesets = None;
651
652        // Propagate tracing context into rayon-spawned threads so that per-segment
653        // write spans appear as children of write_blocks_data in traces.
654        let span = tracing::Span::current();
655        runtime.storage_pool().in_place_scope(|s| {
656            s.spawn(|_| {
657                let _guard = span.enter();
658                r_headers =
659                    Some(self.write_segment(StaticFileSegment::Headers, first_block_number, |w| {
660                        Self::write_headers(w, blocks)
661                    }));
662            });
663
664            s.spawn(|_| {
665                let _guard = span.enter();
666                r_txs = Some(self.write_segment(
667                    StaticFileSegment::Transactions,
668                    first_block_number,
669                    |w| Self::write_transactions(w, blocks, tx_nums),
670                ));
671            });
672
673            if ctx.write_senders {
674                s.spawn(|_| {
675                    let _guard = span.enter();
676                    r_senders = Some(self.write_segment(
677                        StaticFileSegment::TransactionSenders,
678                        first_block_number,
679                        |w| Self::write_transaction_senders(w, blocks, tx_nums),
680                    ));
681                });
682            }
683
684            if ctx.write_receipts {
685                s.spawn(|_| {
686                    let _guard = span.enter();
687                    r_receipts = Some(self.write_segment(
688                        StaticFileSegment::Receipts,
689                        first_block_number,
690                        |w| Self::write_receipts(w, blocks, tx_nums, &ctx),
691                    ));
692                });
693            }
694
695            if ctx.write_account_changesets {
696                s.spawn(|_| {
697                    let _guard = span.enter();
698                    r_account_changesets = Some(self.write_segment(
699                        StaticFileSegment::AccountChangeSets,
700                        first_block_number,
701                        |w| Self::write_account_changesets(w, blocks),
702                    ));
703                });
704            }
705
706            if ctx.write_storage_changesets {
707                s.spawn(|_| {
708                    let _guard = span.enter();
709                    r_storage_changesets = Some(self.write_segment(
710                        StaticFileSegment::StorageChangeSets,
711                        first_block_number,
712                        |w| Self::write_storage_changesets(w, blocks),
713                    ));
714                });
715            }
716        });
717
718        r_headers.ok_or(StaticFileWriterError::ThreadPanic("headers"))??;
719        r_txs.ok_or(StaticFileWriterError::ThreadPanic("transactions"))??;
720        if ctx.write_senders {
721            r_senders.ok_or(StaticFileWriterError::ThreadPanic("senders"))??;
722        }
723        if ctx.write_receipts {
724            r_receipts.ok_or(StaticFileWriterError::ThreadPanic("receipts"))??;
725        }
726        if ctx.write_account_changesets {
727            r_account_changesets
728                .ok_or(StaticFileWriterError::ThreadPanic("account_changesets"))??;
729        }
730        if ctx.write_storage_changesets {
731            r_storage_changesets
732                .ok_or(StaticFileWriterError::ThreadPanic("storage_changesets"))??;
733        }
734        Ok(())
735    }
736
737    /// Gets the [`StaticFileJarProvider`] of the requested segment and start index that can be
738    /// either block or transaction.
739    pub fn get_segment_provider(
740        &self,
741        segment: StaticFileSegment,
742        number: u64,
743    ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
744        if segment.is_block_or_change_based() {
745            self.get_segment_provider_for_block(segment, number, None)
746        } else {
747            self.get_segment_provider_for_transaction(segment, number, None)
748        }
749    }
750
751    /// Gets the [`StaticFileJarProvider`] of the requested segment and start index that can be
752    /// either block or transaction.
753    ///
754    /// If the segment is not found, returns [`None`].
755    pub fn get_maybe_segment_provider(
756        &self,
757        segment: StaticFileSegment,
758        number: u64,
759    ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
760        let provider = if segment.is_block_or_change_based() {
761            self.get_segment_provider_for_block(segment, number, None)
762        } else {
763            self.get_segment_provider_for_transaction(segment, number, None)
764        };
765
766        match provider {
767            Ok(provider) => Ok(Some(provider)),
768            Err(
769                ProviderError::MissingStaticFileBlock(_, _) |
770                ProviderError::MissingStaticFileTx(_, _),
771            ) => Ok(None),
772            Err(err) => Err(err),
773        }
774    }
775
776    /// Gets the [`StaticFileJarProvider`] of the requested segment and block.
777    pub fn get_segment_provider_for_block(
778        &self,
779        segment: StaticFileSegment,
780        block: BlockNumber,
781        path: Option<&Path>,
782    ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
783        self.get_segment_provider_for_range(
784            segment,
785            || self.get_segment_ranges_from_block(segment, block),
786            path,
787        )?
788        .ok_or(ProviderError::MissingStaticFileBlock(segment, block))
789    }
790
791    /// Gets the [`StaticFileJarProvider`] of the requested segment and transaction.
792    pub fn get_segment_provider_for_transaction(
793        &self,
794        segment: StaticFileSegment,
795        tx: TxNumber,
796        path: Option<&Path>,
797    ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
798        self.get_segment_provider_for_range(
799            segment,
800            || self.get_segment_ranges_from_transaction(segment, tx),
801            path,
802        )?
803        .ok_or(ProviderError::MissingStaticFileTx(segment, tx))
804    }
805
806    /// Gets the [`StaticFileJarProvider`] of the requested segment and block or transaction.
807    ///
808    /// `fn_range` should make sure the range goes through `find_fixed_range`.
809    pub fn get_segment_provider_for_range(
810        &self,
811        segment: StaticFileSegment,
812        fn_range: impl Fn() -> Option<SegmentRangeInclusive>,
813        path: Option<&Path>,
814    ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
815        // If we have a path, then get the block range from its name.
816        // Otherwise, check `self.available_static_files`
817        let block_range = match path {
818            Some(path) => StaticFileSegment::parse_filename(
819                &path
820                    .file_name()
821                    .ok_or_else(|| {
822                        ProviderError::MissingStaticFileSegmentPath(segment, path.to_path_buf())
823                    })?
824                    .to_string_lossy(),
825            )
826            .and_then(|(parsed_segment, block_range)| {
827                if parsed_segment == segment {
828                    return Some(block_range);
829                }
830                None
831            }),
832            None => fn_range(),
833        };
834
835        // Return cached `LoadedJar` or insert it for the first time, and then, return it.
836        if let Some(block_range) = block_range {
837            return Ok(Some(self.get_or_create_jar_provider(segment, &block_range)?));
838        }
839
840        Ok(None)
841    }
842
843    /// Gets the [`StaticFileJarProvider`] of the requested path.
844    pub fn get_segment_provider_for_path(
845        &self,
846        path: &Path,
847    ) -> ProviderResult<Option<StaticFileJarProvider<'_, N>>> {
848        StaticFileSegment::parse_filename(
849            &path
850                .file_name()
851                .ok_or_else(|| ProviderError::MissingStaticFilePath(path.to_path_buf()))?
852                .to_string_lossy(),
853        )
854        .map(|(segment, block_range)| self.get_or_create_jar_provider(segment, &block_range))
855        .transpose()
856    }
857
858    /// Given a segment and block range it removes the cached provider from the map.
859    ///
860    /// CAUTION: cached provider should be dropped before calling this or IT WILL deadlock.
861    pub fn remove_cached_provider(
862        &self,
863        segment: StaticFileSegment,
864        fixed_block_range_end: BlockNumber,
865    ) {
866        self.map.remove(&(fixed_block_range_end, segment));
867    }
868
869    /// This handles history expiry by deleting all static files for the given segment below the
870    /// given block.
871    ///
872    /// For example if block is 1M and the blocks per file are 500K this will delete all individual
873    /// files below 1M, so 0-499K and 500K-999K.
874    ///
875    /// This will not delete the file that contains the block itself, because files can only be
876    /// removed entirely.
877    ///
878    /// # Safety
879    ///
880    /// This method will never delete the highest static file for the segment, even if the
881    /// requested block is higher than the highest block in static files. This ensures we always
882    /// maintain at least one static file if any exist.
883    ///
884    /// Returns a list of `SegmentHeader`s from the deleted jars.
885    pub fn delete_segment_below_block(
886        &self,
887        segment: StaticFileSegment,
888        block: BlockNumber,
889    ) -> ProviderResult<Vec<SegmentHeader>> {
890        // Nothing to delete if block is 0.
891        if block == 0 {
892            return Ok(Vec::new());
893        }
894
895        let highest_block = self.get_highest_static_file_block(segment);
896        let mut deleted_headers = Vec::new();
897
898        loop {
899            let Some(block_height) = self.get_lowest_range_end(segment) else {
900                return Ok(deleted_headers);
901            };
902
903            // Stop if we've reached the target block or the highest static file
904            if block_height >= block || Some(block_height) == highest_block {
905                return Ok(deleted_headers);
906            }
907
908            debug!(
909                target: "providers::static_file",
910                ?segment,
911                ?block_height,
912                "Deleting static file below block"
913            );
914
915            // now we need to wipe the static file, this will take care of updating the index and
916            // advance the lowest tracked block height for the segment.
917            let header = self.delete_jar(segment, block_height).inspect_err(|err| {
918                warn!( target: "providers::static_file", ?segment, %block_height, ?err, "Failed to delete static file below block")
919            })?;
920
921            deleted_headers.push(header);
922        }
923    }
924
925    /// Given a segment and block, it deletes the jar and all files from the respective block range.
926    ///
927    /// CAUTION: destructive. Deletes files on disk.
928    ///
929    /// This will re-initialize the index after deletion, so all files are tracked.
930    ///
931    /// Returns the `SegmentHeader` of the deleted jar.
932    pub fn delete_jar(
933        &self,
934        segment: StaticFileSegment,
935        block: BlockNumber,
936    ) -> ProviderResult<SegmentHeader> {
937        let fixed_block_range = self.find_fixed_range(segment, block);
938        let key = (fixed_block_range.end(), segment);
939        let file = self.path.join(segment.filename(&fixed_block_range));
940        let jar = if let Some((_, jar)) = self.map.remove(&key) {
941            jar.jar
942        } else {
943            debug!(
944                target: "providers::static_file",
945                ?file,
946                ?fixed_block_range,
947                ?block,
948                "Loading static file jar for deletion"
949            );
950            NippyJar::<SegmentHeader>::load(&file).map_err(ProviderError::other)?
951        };
952
953        let header = jar.user_header().clone();
954
955        // Delete the sidecar file for changeset segments before deleting the main jar
956        if segment.is_change_based() {
957            let csoff_path = file.with_extension("csoff");
958            if csoff_path.exists() {
959                std::fs::remove_file(&csoff_path).map_err(ProviderError::other)?;
960            }
961        }
962
963        jar.delete().map_err(ProviderError::other)?;
964
965        // SAFETY: this is currently necessary to ensure that certain indexes like
966        // `static_files_min_block` have the correct values after pruning.
967        self.initialize_index()?;
968
969        Ok(header)
970    }
971
972    /// Deletes ALL static file jars for the given segment, including the highest one.
973    ///
974    /// CAUTION: destructive. Deletes all files on disk for this segment.
975    ///
976    /// This is used for `PruneMode::Full` where all data should be removed.
977    ///
978    /// Returns a list of `SegmentHeader`s from the deleted jars.
979    pub fn delete_segment(&self, segment: StaticFileSegment) -> ProviderResult<Vec<SegmentHeader>> {
980        let mut deleted_headers = Vec::new();
981
982        self.writers.remove(segment);
983
984        while let Some(block_height) = self.get_highest_static_file_block(segment) {
985            debug!(
986                target: "providers::static_file",
987                ?segment,
988                ?block_height,
989                "Deleting static file jar"
990            );
991
992            let header = self.delete_jar(segment, block_height).inspect_err(|err| {
993                warn!(target: "providers::static_file", ?segment, %block_height, ?err, "Failed to delete static file jar")
994            })?;
995
996            deleted_headers.push(header);
997        }
998
999        Ok(deleted_headers)
1000    }
1001
1002    /// Given a segment and block range it returns a cached
1003    /// [`StaticFileJarProvider`]. TODO(joshie): we should check the size and pop N if there's too
1004    /// many.
1005    fn get_or_create_jar_provider(
1006        &self,
1007        segment: StaticFileSegment,
1008        fixed_block_range: &SegmentRangeInclusive,
1009    ) -> ProviderResult<StaticFileJarProvider<'_, N>> {
1010        let key = (fixed_block_range.end(), segment);
1011
1012        // Avoid using `entry` directly to avoid a write lock in the common case.
1013        trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Getting provider");
1014        let mut provider: StaticFileJarProvider<'_, N> = if let Some(jar) = self.map.get(&key) {
1015            trace!(target: "providers::static_file", ?segment, ?fixed_block_range, "Jar found in cache");
1016            jar.into()
1017        } else {
1018            let generation = self.cache_generation.load(Ordering::Acquire);
1019            trace!(target: "providers::static_file", ?segment, ?fixed_block_range, generation, "Creating jar from scratch");
1020            let path = self.path.join(segment.filename(fixed_block_range));
1021            let jar = NippyJar::load(&path).map_err(ProviderError::other)?;
1022            let loaded = LoadedJar::new(jar)?;
1023            #[cfg(test)]
1024            self.before_cache_insert.fire(segment);
1025
1026            // The cache may have been populated since the initial miss, including by
1027            // `update_index` publishing a newer snapshot while we loaded this jar without a lock.
1028            // Preserve that entry instead of overwriting it with our potentially stale snapshot.
1029            self.map
1030                .entry(key)
1031                .or_try_insert_with(|| -> ProviderResult<_> {
1032                    // Validate while holding the entry's shard lock so clearing the cache cannot
1033                    // slip between validation and insertion. If invalidated, reload once under
1034                    // this lock rather than retrying unboundedly during frequent pruning.
1035                    let loaded = if self.cache_generation.load(Ordering::Acquire) == generation {
1036                        loaded
1037                    } else {
1038                        trace!(target: "providers::static_file", ?segment, ?fixed_block_range, generation, "Reloading jar after cache invalidation");
1039                        drop(loaded);
1040                        LoadedJar::new(NippyJar::load(&path).map_err(ProviderError::other)?)?
1041                    };
1042                    #[cfg(test)]
1043                    self.after_cache_validation.fire(segment);
1044                    Ok(loaded)
1045                })?
1046                .downgrade()
1047                .into()
1048        };
1049
1050        if let Some(metrics) = &self.metrics {
1051            provider = provider.with_metrics(metrics.clone());
1052        }
1053        Ok(provider)
1054    }
1055
1056    /// Gets a static file segment's block range from the provider inner block
1057    /// index.
1058    fn get_segment_ranges_from_block(
1059        &self,
1060        segment: StaticFileSegment,
1061        block: u64,
1062    ) -> Option<SegmentRangeInclusive> {
1063        let indexes = self.indexes.read();
1064        let index = indexes.get(segment)?;
1065
1066        (index.max_block >= block).then(|| {
1067            self.find_fixed_range_with_block_index(
1068                segment,
1069                Some(&index.expected_block_ranges_by_max_block),
1070                block,
1071            )
1072        })
1073    }
1074
1075    /// Gets a static file segment's fixed block range from the provider inner
1076    /// transaction index.
1077    fn get_segment_ranges_from_transaction(
1078        &self,
1079        segment: StaticFileSegment,
1080        tx: u64,
1081    ) -> Option<SegmentRangeInclusive> {
1082        let indexes = self.indexes.read();
1083        let index = indexes.get(segment)?;
1084        let available_block_ranges_by_max_tx = index.available_block_ranges_by_max_tx.as_ref()?;
1085
1086        // It's more probable that the request comes from a newer tx height, so we iterate
1087        // the static_files in reverse.
1088        let mut static_files_rev_iter = available_block_ranges_by_max_tx.iter().rev().peekable();
1089
1090        while let Some((tx_end, block_range)) = static_files_rev_iter.next() {
1091            if tx > *tx_end {
1092                // request tx is higher than highest static file tx
1093                return None;
1094            }
1095            let tx_start = static_files_rev_iter.peek().map(|(tx_end, _)| *tx_end + 1).unwrap_or(0);
1096            if tx_start <= tx {
1097                return Some(self.find_fixed_range_with_block_index(
1098                    segment,
1099                    Some(&index.expected_block_ranges_by_max_block),
1100                    block_range.end(),
1101                ));
1102            }
1103        }
1104        None
1105    }
1106
1107    /// Updates the inner transaction and block indexes alongside the internal cached providers in
1108    /// `self.map`.
1109    ///
1110    /// Any entry higher than `segment_max_block` will be deleted from the previous structures.
1111    ///
1112    /// If `segment_max_block` is None it means there's no static file for this segment.
1113    pub fn update_index(
1114        &self,
1115        segment: StaticFileSegment,
1116        segment_max_block: Option<BlockNumber>,
1117    ) -> ProviderResult<()> {
1118        trace!(
1119            target: "providers::static_file",
1120            ?segment,
1121            ?segment_max_block,
1122            "Updating provider index"
1123        );
1124        let mut indexes = self.indexes.write();
1125
1126        match segment_max_block {
1127            Some(segment_max_block) => {
1128                let fixed_range = self.find_fixed_range_with_block_index(
1129                    segment,
1130                    indexes.get(segment).map(|index| &index.expected_block_ranges_by_max_block),
1131                    segment_max_block,
1132                );
1133
1134                let jar = NippyJar::<SegmentHeader>::load(
1135                    &self.path.join(segment.filename(&fixed_range)),
1136                )
1137                .map_err(ProviderError::other)?;
1138
1139                let index = indexes
1140                    .entry(segment)
1141                    .and_modify(|index| {
1142                        // Update max block
1143                        index.max_block = segment_max_block;
1144
1145                        // Update expected block range index
1146
1147                        // Remove all expected block ranges that are less than the new max block
1148                        index
1149                            .expected_block_ranges_by_max_block
1150                            .retain(|_, block_range| block_range.start() < fixed_range.start());
1151                        // Insert new expected block range
1152                        index
1153                            .expected_block_ranges_by_max_block
1154                            .insert(fixed_range.end(), fixed_range);
1155                    })
1156                    .or_insert_with(|| StaticFileSegmentIndex {
1157                        min_block_range: None,
1158                        max_block: segment_max_block,
1159                        expected_block_ranges_by_max_block: BTreeMap::from([(
1160                            fixed_range.end(),
1161                            fixed_range,
1162                        )]),
1163                        available_block_ranges_by_max_tx: None,
1164                    });
1165
1166                // Update min_block to track the lowest block range of the segment.
1167                // This is initially set by initialize_index() on node startup, but must be updated
1168                // as the file grows to prevent stale values.
1169                //
1170                // Without this update, min_block can remain at genesis (e.g. Some([0..=0]) or None)
1171                // even after syncing to higher blocks (e.g. [0..=100]). A stale
1172                // min_block causes get_lowest_static_file_block() to return the
1173                // wrong end value, which breaks pruning logic that relies on it for
1174                // safety checks.
1175                //
1176                // Example progression:
1177                // 1. Node starts, initialize_index() sets min_block = [0..=0]
1178                // 2. Sync to block 100, this update sets min_block = [0..=100]
1179                // 3. Pruner calls get_lowest_static_file_block() -> returns 100 (correct). Without
1180                //    this update, it would incorrectly return 0 (stale)
1181                if let Some(current_block_range) = jar.user_header().block_range() {
1182                    if let Some(min_block_range) = index.min_block_range.as_mut() {
1183                        // delete_jar WILL ALWAYS re-initialize all indexes, so we are always
1184                        // sure that current_min is always the lowest.
1185                        if current_block_range.start() == min_block_range.start() {
1186                            *min_block_range = current_block_range;
1187                        }
1188                    } else {
1189                        index.min_block_range = Some(current_block_range);
1190                    }
1191                }
1192
1193                // Updates the tx index by first removing all entries which have a higher
1194                // block_start than our current static file.
1195                if let Some(tx_range) = jar.user_header().tx_range() {
1196                    // Current block range has the same block start as `fixed_range``, but block end
1197                    // might be different if we are still filling this static file.
1198                    if let Some(current_block_range) = jar.user_header().block_range() {
1199                        let tx_end = tx_range.end();
1200
1201                        // Considering that `update_index` is called when we either append/truncate,
1202                        // we are sure that we are handling the latest data
1203                        // points.
1204                        //
1205                        // Here we remove every entry of the index that has a block start higher or
1206                        // equal than our current one. This is important in the case
1207                        // that we prune a lot of rows resulting in a file (and thus
1208                        // a higher block range) deletion.
1209                        if let Some(index) = index.available_block_ranges_by_max_tx.as_mut() {
1210                            index
1211                                .retain(|_, block_range| block_range.start() < fixed_range.start());
1212                            index.insert(tx_end, current_block_range);
1213                        } else {
1214                            index.available_block_ranges_by_max_tx =
1215                                Some(BTreeMap::from([(tx_end, current_block_range)]));
1216                        }
1217                    }
1218                } else if segment.is_tx_based() {
1219                    // The unwinded file has no more transactions/receipts. However, the highest
1220                    // block is within this files' block range. We only retain
1221                    // entries with block ranges before the current one.
1222                    if let Some(index) = index.available_block_ranges_by_max_tx.as_mut() {
1223                        index.retain(|_, block_range| block_range.start() < fixed_range.start());
1224                    }
1225
1226                    // If the index is empty, just remove it.
1227                    index.available_block_ranges_by_max_tx.take_if(|index| index.is_empty());
1228                }
1229
1230                // Update the cached provider.
1231                trace!(target: "providers::static_file", ?segment, "Inserting updated jar into cache");
1232                self.map.insert((fixed_range.end(), segment), LoadedJar::new(jar)?);
1233
1234                // Delete any cached provider that no longer has an associated jar.
1235                trace!(target: "providers::static_file", ?segment, "Cleaning up jar map");
1236                self.map.retain(|(end, seg), _| !(*seg == segment && *end > fixed_range.end()));
1237            }
1238            None => {
1239                debug!(target: "providers::static_file", ?segment, "Removing segment from index");
1240                indexes.remove(segment);
1241            }
1242        };
1243
1244        trace!(target: "providers::static_file", ?segment, "Updated provider index");
1245        Ok(())
1246    }
1247
1248    /// Initializes the inner transaction and block index
1249    pub fn initialize_index(&self) -> ProviderResult<()> {
1250        let mut indexes = self.indexes.write();
1251        indexes.clear();
1252
1253        for (segment, headers) in &*iter_static_files(&self.path).map_err(ProviderError::other)? {
1254            // Update first and last block for each segment
1255            //
1256            // It's safe to call `expect` here, because every segment has at least one header
1257            // associated with it.
1258            let min_block_range = Some(headers.first().expect("headers are not empty").0);
1259            let max_block = headers.last().expect("headers are not empty").0.end();
1260
1261            let mut expected_block_ranges_by_max_block = BTreeMap::default();
1262            let mut available_block_ranges_by_max_tx = None;
1263
1264            for (block_range, header) in headers {
1265                // Update max expected block -> expected_block_range index
1266                expected_block_ranges_by_max_block
1267                    .insert(header.expected_block_end(), header.expected_block_range());
1268
1269                // Update max tx -> block_range index
1270                if let Some(tx_range) = header.tx_range() {
1271                    let tx_end = tx_range.end();
1272
1273                    available_block_ranges_by_max_tx
1274                        .get_or_insert_with(BTreeMap::default)
1275                        .insert(tx_end, *block_range);
1276                }
1277            }
1278
1279            indexes.insert(
1280                segment,
1281                StaticFileSegmentIndex {
1282                    min_block_range,
1283                    max_block,
1284                    expected_block_ranges_by_max_block,
1285                    available_block_ranges_by_max_tx,
1286                },
1287            );
1288        }
1289
1290        // If this is a re-initialization, invalidate in-flight cache fills before clearing the
1291        // cache. A fill that started before this point either reloads under its shard lock or
1292        // publishes before the clear reaches that shard and is removed by it.
1293        self.cache_generation.fetch_add(1, Ordering::AcqRel);
1294        self.map.clear();
1295
1296        // initialize the expired history height to the lowest static file block
1297        if let Some(lowest_range) =
1298            indexes.get(StaticFileSegment::Transactions).and_then(|index| index.min_block_range)
1299        {
1300            // the earliest height is the lowest available block number
1301            self.earliest_history_height
1302                .store(lowest_range.start(), std::sync::atomic::Ordering::Relaxed);
1303        }
1304
1305        Ok(())
1306    }
1307
1308    /// Ensures that any broken invariants which cannot be healed on the spot return a pipeline
1309    /// target to unwind to.
1310    ///
1311    /// Two types of consistency checks are done for:
1312    ///
1313    /// 1) When a static file fails to commit but the underlying data was changed.
1314    /// 2) When a static file was committed, but the required database transaction was not.
1315    ///
1316    /// For 1) it can self-heal if `self.access.is_read_only()` is set to `false`. Otherwise, it
1317    /// will return an error.
1318    /// For 2) the invariants below are checked, and if broken, might require a pipeline unwind
1319    /// to heal.
1320    ///
1321    /// For each static file segment:
1322    /// * the corresponding database table should overlap or have continuity in their keys
1323    ///   ([`TxNumber`] or [`BlockNumber`]).
1324    /// * its highest block should match the stage checkpoint block number if it's equal or higher
1325    ///   than the corresponding database table last entry.
1326    ///
1327    /// Returns a [`Option`] of [`PipelineTarget::Unwind`] if any healing is further required.
1328    ///
1329    /// WARNING: No static file writer should be held before calling this function, otherwise it
1330    /// will deadlock.
1331    #[instrument(skip(self, provider), fields(read_only = self.is_read_only()))]
1332    pub fn check_consistency<Provider>(
1333        &self,
1334        provider: &Provider,
1335    ) -> ProviderResult<Option<PipelineTarget>>
1336    where
1337        Provider: DBProvider
1338            + BlockReader
1339            + StageCheckpointReader
1340            + PruneCheckpointReader
1341            + ChainSpecProvider
1342            + StorageSettingsCache,
1343        N: NodePrimitives<Receipt: Value, BlockHeader: Value, SignedTx: Value>,
1344    {
1345        // OVM historical import is broken and does not work with this check. It's importing
1346        // duplicated receipts resulting in having more receipts than the expected transaction
1347        // range.
1348        //
1349        // If we detect an OVM import was done (block #1 <https://optimistic.etherscan.io/block/1>), skip it.
1350        // More on [#11099](https://github.com/paradigmxyz/reth/pull/11099).
1351        if provider.chain_spec().is_optimism() &&
1352            reth_chainspec::Chain::optimism_mainnet() == provider.chain_spec().chain_id()
1353        {
1354            // check whether we have the first OVM block: <https://optimistic.etherscan.io/block/0xbee7192e575af30420cae0c7776304ac196077ee72b048970549e4f08e875453>
1355            const OVM_HEADER_1_HASH: B256 =
1356                b256!("0xbee7192e575af30420cae0c7776304ac196077ee72b048970549e4f08e875453");
1357            if provider.block_number(OVM_HEADER_1_HASH)?.is_some() {
1358                info!(target: "reth::cli",
1359                    "Skipping storage verification for OP mainnet, expected inconsistency in OVM chain"
1360                );
1361                return Ok(None);
1362            }
1363        }
1364
1365        info!(target: "reth::cli", "Verifying storage consistency.");
1366
1367        let mut unwind_target: Option<BlockNumber> = None;
1368
1369        let mut update_unwind_target = |new_target| {
1370            unwind_target =
1371                unwind_target.map(|current| current.min(new_target)).or(Some(new_target));
1372        };
1373
1374        for segment in self.segments_to_check(provider) {
1375            let span = info_span!(
1376                "Checking consistency for segment",
1377                ?segment,
1378                initial_highest_block = tracing::field::Empty,
1379                highest_block = tracing::field::Empty,
1380                highest_tx = tracing::field::Empty,
1381            );
1382            let _guard = span.enter();
1383
1384            debug!(target: "reth::providers::static_file", "Checking consistency for segment");
1385
1386            // Heal file-level inconsistencies and get before/after highest block
1387            let (initial_highest_block, mut highest_block) = self.maybe_heal_segment(segment)?;
1388            span.record("initial_highest_block", initial_highest_block);
1389            span.record("highest_block", highest_block);
1390
1391            // Only applies to block-based static files. (Headers)
1392            //
1393            // The updated `highest_block` may have decreased if we healed from a pruning
1394            // interruption.
1395            if initial_highest_block != highest_block {
1396                info!(
1397                    target: "reth::providers::static_file",
1398                    unwind_target = highest_block,
1399                    "Setting unwind target."
1400                );
1401                update_unwind_target(highest_block.unwrap_or_default());
1402            }
1403
1404            // Only applies to transaction-based static files. (Receipts & Transactions)
1405            //
1406            // Make sure the last transaction matches the last block from its indices, since a heal
1407            // from a pruning interruption might have decreased the number of transactions without
1408            // being able to update the last block of the static file segment.
1409            let highest_tx = self.get_highest_static_file_tx(segment);
1410            span.record("highest_tx", highest_tx);
1411            debug!(target: "reth::providers::static_file", "Checking tx index segment");
1412
1413            if let Some(highest_tx) = highest_tx {
1414                let mut last_block = highest_block.unwrap_or_default();
1415                debug!(target: "reth::providers::static_file", last_block, highest_tx, "Verifying last transaction matches last block indices");
1416                loop {
1417                    let Some(indices) = provider.block_body_indices(last_block)? else {
1418                        debug!(target: "reth::providers::static_file", last_block, "Block body indices not found, static files ahead of database");
1419                        // If the block body indices can not be found, then it means that static
1420                        // files is ahead of database, and the `ensure_invariants` check will fix
1421                        // it by comparing with stage checkpoints.
1422                        break
1423                    };
1424
1425                    debug!(target: "reth::providers::static_file", last_block, last_tx_num = indices.last_tx_num(), "Found block body indices");
1426
1427                    if indices.last_tx_num() <= highest_tx {
1428                        break
1429                    }
1430
1431                    if last_block == 0 {
1432                        debug!(target: "reth::providers::static_file", "Reached block 0 in verification loop");
1433                        break
1434                    }
1435
1436                    last_block -= 1;
1437
1438                    info!(
1439                        target: "reth::providers::static_file",
1440                        highest_block = self.get_highest_static_file_block(segment),
1441                        unwind_target = last_block,
1442                        "Setting unwind target."
1443                    );
1444                    span.record("highest_block", last_block);
1445                    highest_block = Some(last_block);
1446                    update_unwind_target(last_block);
1447                }
1448            }
1449
1450            debug!(target: "reth::providers::static_file", "Ensuring invariants for segment");
1451
1452            match self.ensure_invariants_for(provider, segment, highest_tx, highest_block)? {
1453                Some(unwind) => {
1454                    debug!(target: "reth::providers::static_file", unwind_target=unwind, "Invariants check returned unwind target");
1455                    update_unwind_target(unwind);
1456                }
1457                None => {
1458                    debug!(target: "reth::providers::static_file", "Invariants check completed, no unwind needed")
1459                }
1460            }
1461        }
1462
1463        Ok(unwind_target.map(PipelineTarget::Unwind))
1464    }
1465
1466    /// Heals file-level (`NippyJar`) inconsistencies for eligible static file
1467    /// segments.
1468    ///
1469    /// Call before [`Self::check_consistency`] so files are internally
1470    /// consistent.
1471    ///
1472    /// Uses the same segment-skip logic as [`Self::check_consistency`], but
1473    /// does not compare with database checkpoints or prune against them.
1474    pub fn check_file_consistency<Provider>(&self, provider: &Provider) -> ProviderResult<()>
1475    where
1476        Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1477    {
1478        info!(target: "reth::cli", "Healing static file inconsistencies.");
1479
1480        for segment in self.segments_to_check(provider) {
1481            let _guard = info_span!("healing_static_file_segment", ?segment).entered();
1482            let _ = self.maybe_heal_segment(segment)?;
1483        }
1484
1485        Ok(())
1486    }
1487
1488    /// Returns the static file segments that should be checked/healed for this provider.
1489    fn segments_to_check<'a, Provider>(
1490        &'a self,
1491        provider: &'a Provider,
1492    ) -> impl Iterator<Item = StaticFileSegment> + 'a
1493    where
1494        Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1495    {
1496        StaticFileSegment::iter()
1497            .filter(move |segment| self.should_check_segment(provider, *segment))
1498    }
1499
1500    /// True if the given segment should be checked/healed for this provider.
1501    fn should_check_segment<Provider>(
1502        &self,
1503        provider: &Provider,
1504        segment: StaticFileSegment,
1505    ) -> bool
1506    where
1507        Provider: DBProvider + ChainSpecProvider + StorageSettingsCache + PruneCheckpointReader,
1508    {
1509        match segment {
1510            StaticFileSegment::Headers | StaticFileSegment::Transactions => true,
1511            StaticFileSegment::Receipts => {
1512                if EitherWriter::receipts_destination(provider).is_database() {
1513                    // Old pruned nodes (including full node) do not store receipts as static
1514                    // files.
1515                    debug!(target: "reth::providers::static_file", ?segment, "Skipping receipts segment: receipts stored in database");
1516                    return false;
1517                }
1518
1519                if NamedChain::Gnosis == provider.chain_spec().chain_id() ||
1520                    NamedChain::Chiado == provider.chain_spec().chain_id()
1521                {
1522                    // Gnosis and Chiado's historical import is broken and does not work with
1523                    // this check. They are importing receipts along
1524                    // with importing headers/bodies.
1525                    debug!(target: "reth::providers::static_file", ?segment, "Skipping receipts segment: broken historical import for gnosis/chiado");
1526                    return false;
1527                }
1528
1529                true
1530            }
1531            StaticFileSegment::TransactionSenders => {
1532                if EitherWriterDestination::senders(provider).is_database() {
1533                    debug!(target: "reth::providers::static_file", ?segment, "Skipping senders segment: senders stored in database");
1534                    return false;
1535                }
1536
1537                if Self::is_segment_fully_pruned(provider, PruneSegment::SenderRecovery) {
1538                    debug!(target: "reth::providers::static_file", ?segment, "Skipping senders segment: fully pruned");
1539                    return false;
1540                }
1541
1542                true
1543            }
1544            StaticFileSegment::AccountChangeSets => {
1545                if EitherWriter::account_changesets_destination(provider).is_database() {
1546                    debug!(target: "reth::providers::static_file", ?segment, "Skipping account changesets segment: changesets stored in database");
1547                    return false;
1548                }
1549                true
1550            }
1551            StaticFileSegment::StorageChangeSets => {
1552                if EitherWriter::storage_changesets_destination(provider).is_database() {
1553                    debug!(target: "reth::providers::static_file", ?segment, "Skipping storage changesets segment: changesets stored in database");
1554                    return false
1555                }
1556                true
1557            }
1558        }
1559    }
1560
1561    /// Returns `true` if the given prune segment has a checkpoint with
1562    /// [`reth_prune_types::PruneMode::Full`], indicating all data for this segment has been
1563    /// intentionally deleted.
1564    fn is_segment_fully_pruned<Provider>(provider: &Provider, segment: PruneSegment) -> bool
1565    where
1566        Provider: PruneCheckpointReader,
1567    {
1568        provider
1569            .get_prune_checkpoint(segment)
1570            .ok()
1571            .flatten()
1572            .is_some_and(|checkpoint| checkpoint.prune_mode.is_full())
1573    }
1574
1575    /// Returns the first transaction number after `block`, or `None` if it can't be derived.
1576    ///
1577    /// Falls back to the `Bodies` prune checkpoint when bodies through `block` were never stored,
1578    /// e.g. below a snap sync pivot, where transaction numbers restart at 0.
1579    fn next_tx_num_after<Provider>(
1580        provider: &Provider,
1581        block: BlockNumber,
1582    ) -> ProviderResult<Option<TxNumber>>
1583    where
1584        Provider: BlockReader + PruneCheckpointReader,
1585    {
1586        if let Some(indices) = provider.block_body_indices(block)? {
1587            return Ok(Some(indices.next_tx_num()))
1588        }
1589        Ok(provider
1590            .get_prune_checkpoint(PruneSegment::Bodies)?
1591            .filter(|pruned| {
1592                pruned.tx_number.is_none() &&
1593                    pruned.block_number.is_some_and(|pruned| pruned >= block)
1594            })
1595            .map(|_| 0))
1596    }
1597
1598    /// Checks consistency of the latest static file segment and throws an
1599    /// error if at fault.
1600    ///
1601    /// Read-only.
1602    fn check_segment_consistency(&self, segment: StaticFileSegment) -> ProviderResult<()> {
1603        debug!(target: "reth::providers::static_file", "Checking segment consistency");
1604        if let Some(latest_block) = self.get_highest_static_file_block(segment) {
1605            let file_path = self
1606                .directory()
1607                .join(segment.filename(&self.find_fixed_range(segment, latest_block)));
1608            debug!(target: "reth::providers::static_file", ?file_path, latest_block, "Loading NippyJar for consistency check");
1609
1610            let jar = NippyJar::<SegmentHeader>::load(&file_path).map_err(ProviderError::other)?;
1611            debug!(target: "reth::providers::static_file", "NippyJar loaded, checking consistency");
1612
1613            NippyJarChecker::new(jar).check_consistency().map_err(ProviderError::other)?;
1614            debug!(target: "reth::providers::static_file", "NippyJar consistency check passed");
1615        } else {
1616            debug!(target: "reth::providers::static_file", "No static file block found, skipping consistency check");
1617        }
1618        Ok(())
1619    }
1620
1621    /// Attempts to heal file-level (`NippyJar`) inconsistencies for a single static file segment.
1622    ///
1623    /// Returns the highest block before and after healing, which can be used to detect
1624    /// if healing from a pruning interruption decreased the highest block.
1625    ///
1626    /// File consistency is broken if:
1627    ///
1628    /// * appending data was interrupted before a config commit, then data file will be truncated
1629    ///   according to the config.
1630    ///
1631    /// * pruning data was interrupted before a config commit, then we have deleted data that we are
1632    ///   expected to still have. We need to check the Database and unwind everything accordingly.
1633    ///
1634    /// **Note:** In read-only mode, this will return an error if a consistency issue is detected,
1635    /// since healing requires write access.
1636    fn maybe_heal_segment(
1637        &self,
1638        segment: StaticFileSegment,
1639    ) -> ProviderResult<(Option<BlockNumber>, Option<BlockNumber>)> {
1640        let initial_highest_block = self.get_highest_static_file_block(segment);
1641        debug!(target: "reth::providers::static_file", ?initial_highest_block, "Initial highest block for segment");
1642
1643        if self.access.is_read_only() {
1644            // Read-only mode: cannot modify files, so just validate consistency and error if
1645            // broken.
1646            debug!(target: "reth::providers::static_file", "Checking segment consistency (read-only)");
1647            self.check_segment_consistency(segment)?;
1648        } else {
1649            // Writable mode: fetching the writer will automatically heal any file-level
1650            // inconsistency by truncating data to match the last committed config.
1651            debug!(target: "reth::providers::static_file", "Fetching latest writer which might heal any potential inconsistency");
1652            self.latest_writer(segment)?;
1653        }
1654
1655        // The updated `highest_block` may have decreased if we healed from a
1656        // pruning interruption.
1657        let highest_block = self.get_highest_static_file_block(segment);
1658
1659        Ok((initial_highest_block, highest_block))
1660    }
1661
1662    /// Ensure invariants for each corresponding table and static file segment.
1663    fn ensure_invariants_for<Provider>(
1664        &self,
1665        provider: &Provider,
1666        segment: StaticFileSegment,
1667        highest_tx: Option<u64>,
1668        highest_block: Option<BlockNumber>,
1669    ) -> ProviderResult<Option<BlockNumber>>
1670    where
1671        Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1672        N: NodePrimitives<Receipt: Value, BlockHeader: Value, SignedTx: Value>,
1673    {
1674        match segment {
1675            StaticFileSegment::Headers => self
1676                .ensure_invariants::<_, tables::Headers<N::BlockHeader>>(
1677                    provider,
1678                    segment,
1679                    highest_block,
1680                    highest_block,
1681                ),
1682            StaticFileSegment::Transactions => self
1683                .ensure_invariants::<_, tables::Transactions<N::SignedTx>>(
1684                    provider,
1685                    segment,
1686                    highest_tx,
1687                    highest_block,
1688                ),
1689            StaticFileSegment::Receipts => self
1690                .ensure_invariants::<_, tables::Receipts<N::Receipt>>(
1691                    provider,
1692                    segment,
1693                    highest_tx,
1694                    highest_block,
1695                ),
1696            StaticFileSegment::TransactionSenders => self
1697                .ensure_invariants::<_, tables::TransactionSenders>(
1698                    provider,
1699                    segment,
1700                    highest_tx,
1701                    highest_block,
1702                ),
1703            StaticFileSegment::AccountChangeSets => self
1704                .ensure_invariants::<_, tables::AccountChangeSets>(
1705                    provider,
1706                    segment,
1707                    highest_tx,
1708                    highest_block,
1709                ),
1710            StaticFileSegment::StorageChangeSets => self
1711                .ensure_changeset_invariants_by_block::<_, tables::StorageChangeSets, _>(
1712                    provider,
1713                    segment,
1714                    highest_block,
1715                    |key| key.block_number(),
1716                ),
1717        }
1718    }
1719
1720    /// Check invariants for each corresponding table and static file segment:
1721    ///
1722    /// * the corresponding database table should overlap or have continuity in their keys
1723    ///   ([`TxNumber`] or [`BlockNumber`]).
1724    /// * its highest block should match the stage checkpoint block number if it's equal or higher
1725    ///   than the corresponding database table last entry.
1726    ///   * If the checkpoint block is higher, then request a pipeline unwind to the static file
1727    ///     block. This is expressed by returning [`Some`] with the requested pipeline unwind
1728    ///     target.
1729    ///   * If the checkpoint block is lower, then heal by removing rows from the static file. In
1730    ///     this case, the rows will be removed and [`None`] will be returned.
1731    ///
1732    /// * If the database tables overlap with static files and have contiguous keys, or the
1733    ///   checkpoint block matches the highest static files block, then [`None`] will be returned.
1734    #[instrument(skip(self, provider, segment), fields(table = T::NAME))]
1735    fn ensure_invariants<Provider, T: Table<Key = u64>>(
1736        &self,
1737        provider: &Provider,
1738        segment: StaticFileSegment,
1739        highest_static_file_entry: Option<u64>,
1740        highest_static_file_block: Option<BlockNumber>,
1741    ) -> ProviderResult<Option<BlockNumber>>
1742    where
1743        Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1744    {
1745        debug!(target: "reth::providers::static_file", "Ensuring invariants");
1746        let mut db_cursor = provider.tx_ref().cursor_read::<T>()?;
1747
1748        if let Some((db_first_entry, _)) = db_cursor.first()? {
1749            debug!(target: "reth::providers::static_file", db_first_entry, "Found first database entry");
1750            if let (Some(highest_entry), Some(highest_block)) =
1751                (highest_static_file_entry, highest_static_file_block)
1752            {
1753                // If there is a gap between the entry found in static file and
1754                // database, then we have most likely lost static file data and need to unwind so we
1755                // can load it again
1756                if !(db_first_entry <= highest_entry || highest_entry + 1 == db_first_entry) {
1757                    info!(
1758                        target: "reth::providers::static_file",
1759                        ?db_first_entry,
1760                        ?highest_entry,
1761                        unwind_target = highest_block,
1762                        "Setting unwind target."
1763                    );
1764                    return Ok(Some(highest_block));
1765                }
1766            }
1767
1768            if let Some((db_last_entry, _)) = db_cursor.last()? &&
1769                highest_static_file_entry
1770                    .is_none_or(|highest_entry| db_last_entry > highest_entry)
1771            {
1772                debug!(target: "reth::providers::static_file", db_last_entry, "Database has entries beyond static files, no unwind needed");
1773                return Ok(None)
1774            }
1775        } else {
1776            debug!(target: "reth::providers::static_file", "No database entries found");
1777        }
1778
1779        let highest_static_file_entry = highest_static_file_entry.unwrap_or_default();
1780        let highest_static_file_block = highest_static_file_block.unwrap_or_default();
1781
1782        // If static file entry is ahead of the database entries, then ensure the checkpoint block
1783        // number matches.
1784        let stage_id = segment.to_stage_id();
1785        let checkpoint_block_number =
1786            provider.get_stage_checkpoint(stage_id)?.unwrap_or_default().block_number;
1787        debug!(target: "reth::providers::static_file", ?stage_id, checkpoint_block_number, "Retrieved stage checkpoint");
1788
1789        let effective_coverage_block =
1790            Self::effective_coverage_block(provider, segment, highest_static_file_block)?;
1791
1792        // If the checkpoint is ahead, then we lost static file data. May be data corruption.
1793        if checkpoint_block_number > effective_coverage_block {
1794            info!(
1795                target: "reth::providers::static_file",
1796                checkpoint_block_number,
1797                unwind_target = effective_coverage_block,
1798                "Setting unwind target."
1799            );
1800            return Ok(Some(effective_coverage_block));
1801        }
1802
1803        // If the checkpoint is ahead, or matches, then nothing to do.
1804        if checkpoint_block_number >= highest_static_file_block {
1805            debug!(target: "reth::providers::static_file", "Invariants ensured, returning None");
1806            return Ok(None);
1807        }
1808
1809        // If the checkpoint is behind, then we failed to do a database commit
1810        // **but committed** to static files on executing a stage, or the
1811        // reverse on unwinding a stage.
1812        //
1813        // All we need to do is to prune the extra static file rows.
1814        info!(
1815            target: "reth::providers",
1816            from = highest_static_file_block,
1817            to = checkpoint_block_number,
1818            "Unwinding static file segment."
1819        );
1820        let mut writer = self.latest_writer(segment)?;
1821
1822        match segment {
1823            StaticFileSegment::Headers => {
1824                let prune_count = highest_static_file_block - checkpoint_block_number;
1825                debug!(target: "reth::providers::static_file", prune_count, "Pruning headers");
1826                // TODO(joshie): is_block_meta
1827                writer.prune_headers(prune_count)?;
1828            }
1829            StaticFileSegment::Transactions |
1830            StaticFileSegment::Receipts |
1831            StaticFileSegment::TransactionSenders => {
1832                if let Some(next_tx_num) =
1833                    Self::next_tx_num_after(provider, checkpoint_block_number)?
1834                {
1835                    // `last_tx_num()` saturates to zero for an empty genesis block, but row zero
1836                    // belongs to the first non-empty block and must be removed as well.
1837                    let number =
1838                        highest_static_file_entry.saturating_add(1).saturating_sub(next_tx_num);
1839                    debug!(target: "reth::providers::static_file", prune_count = number, checkpoint_block_number, "Pruning transaction based segment");
1840
1841                    match segment {
1842                        StaticFileSegment::Transactions => {
1843                            writer.prune_transactions(number, checkpoint_block_number)?
1844                        }
1845                        StaticFileSegment::Receipts => {
1846                            writer.prune_receipts(number, checkpoint_block_number)?
1847                        }
1848                        StaticFileSegment::TransactionSenders => {
1849                            writer.prune_transaction_senders(number, checkpoint_block_number)?
1850                        }
1851                        StaticFileSegment::Headers |
1852                        StaticFileSegment::AccountChangeSets |
1853                        StaticFileSegment::StorageChangeSets => {
1854                            unreachable!()
1855                        }
1856                    }
1857                } else {
1858                    debug!(target: "reth::providers::static_file", checkpoint_block_number, "No block body indices found for checkpoint block");
1859                }
1860            }
1861            StaticFileSegment::AccountChangeSets => {
1862                writer.prune_account_changesets(checkpoint_block_number)?;
1863            }
1864            StaticFileSegment::StorageChangeSets => {
1865                writer.prune_storage_changesets(checkpoint_block_number)?;
1866            }
1867        }
1868
1869        debug!(target: "reth::providers::static_file", "Committing writer after pruning");
1870        writer.commit()?;
1871        debug!(target: "reth::providers::static_file", "Writer committed successfully");
1872
1873        debug!(target: "reth::providers::static_file", "Invariants ensured, returning None");
1874        Ok(None)
1875    }
1876
1877    fn ensure_changeset_invariants_by_block<Provider, T, F>(
1878        &self,
1879        provider: &Provider,
1880        segment: StaticFileSegment,
1881        highest_static_file_block: Option<BlockNumber>,
1882        block_from_key: F,
1883    ) -> ProviderResult<Option<BlockNumber>>
1884    where
1885        Provider: DBProvider + BlockReader + StageCheckpointReader + PruneCheckpointReader,
1886        T: Table,
1887        F: Fn(&T::Key) -> BlockNumber,
1888    {
1889        debug!(
1890            target: "reth::providers::static_file",
1891            ?segment,
1892            ?highest_static_file_block,
1893            "Ensuring changeset invariants"
1894        );
1895        let mut db_cursor = provider.tx_ref().cursor_read::<T>()?;
1896
1897        if let Some((db_first_key, _)) = db_cursor.first()? {
1898            let db_first_block = block_from_key(&db_first_key);
1899            if let Some(highest_block) = highest_static_file_block &&
1900                !(db_first_block <= highest_block || highest_block + 1 == db_first_block)
1901            {
1902                info!(
1903                    target: "reth::providers::static_file",
1904                    ?db_first_block,
1905                    ?highest_block,
1906                    unwind_target = highest_block,
1907                    ?segment,
1908                    "Setting unwind target."
1909                );
1910                return Ok(Some(highest_block))
1911            }
1912
1913            if let Some((db_last_key, _)) = db_cursor.last()? &&
1914                highest_static_file_block
1915                    .is_none_or(|highest_block| block_from_key(&db_last_key) > highest_block)
1916            {
1917                debug!(
1918                    target: "reth::providers::static_file",
1919                    ?segment,
1920                    "Database has entries beyond static files, no unwind needed"
1921                );
1922                return Ok(None)
1923            }
1924        } else {
1925            debug!(target: "reth::providers::static_file", ?segment, "No database entries found");
1926        }
1927
1928        let highest_static_file_block = highest_static_file_block.unwrap_or_default();
1929
1930        let stage_id = segment.to_stage_id();
1931        let checkpoint_block_number =
1932            provider.get_stage_checkpoint(stage_id)?.unwrap_or_default().block_number;
1933
1934        let effective_coverage_block =
1935            Self::effective_coverage_block(provider, segment, highest_static_file_block)?;
1936
1937        if checkpoint_block_number > effective_coverage_block {
1938            info!(
1939                target: "reth::providers::static_file",
1940                checkpoint_block_number,
1941                unwind_target = effective_coverage_block,
1942                ?segment,
1943                "Setting unwind target."
1944            );
1945            return Ok(Some(effective_coverage_block))
1946        }
1947
1948        if checkpoint_block_number < highest_static_file_block {
1949            info!(
1950                target: "reth::providers",
1951                ?segment,
1952                from = highest_static_file_block,
1953                to = checkpoint_block_number,
1954                "Unwinding static file segment."
1955            );
1956            let mut writer = self.latest_writer(segment)?;
1957            match segment {
1958                StaticFileSegment::AccountChangeSets => {
1959                    writer.prune_account_changesets(checkpoint_block_number)?;
1960                }
1961                StaticFileSegment::StorageChangeSets => {
1962                    writer.prune_storage_changesets(checkpoint_block_number)?;
1963                }
1964                _ => unreachable!("invalid segment for changeset invariants"),
1965            }
1966            writer.commit()?;
1967        }
1968
1969        Ok(None)
1970    }
1971
1972    /// Returns the highest block accounted for in this segment, either through data
1973    /// present in static files or through data intentionally removed by pruning.
1974    ///
1975    /// Data below a segment's prune checkpoint has been intentionally deleted, so its
1976    /// absence from static files is not an inconsistency. Without this, a pruned segment
1977    /// whose stage checkpoint is ahead of its (empty) static files is treated as data
1978    /// corruption and triggers an unwind to block 0, which aborts the node on startup.
1979    /// See <https://github.com/paradigmxyz/reth/issues/23463>.
1980    fn effective_coverage_block<Provider>(
1981        provider: &Provider,
1982        segment: StaticFileSegment,
1983        highest_static_file_block: BlockNumber,
1984    ) -> ProviderResult<BlockNumber>
1985    where
1986        Provider: PruneCheckpointReader,
1987    {
1988        let Some(prune_segment) = Self::prune_segment_for_static_file(segment) else {
1989            return Ok(highest_static_file_block)
1990        };
1991
1992        let prune_checkpoint_block = provider
1993            .get_prune_checkpoint(prune_segment)?
1994            .and_then(|checkpoint| checkpoint.block_number)
1995            .unwrap_or_default();
1996
1997        Ok(highest_static_file_block.max(prune_checkpoint_block))
1998    }
1999
2000    /// Returns the prune segment that governs data availability for a static file segment,
2001    /// or `None` if the segment is never pruned.
2002    const fn prune_segment_for_static_file(segment: StaticFileSegment) -> Option<PruneSegment> {
2003        match segment {
2004            StaticFileSegment::Receipts => Some(PruneSegment::Receipts),
2005            StaticFileSegment::TransactionSenders => Some(PruneSegment::SenderRecovery),
2006            StaticFileSegment::AccountChangeSets => Some(PruneSegment::AccountHistory),
2007            StaticFileSegment::StorageChangeSets => Some(PruneSegment::StorageHistory),
2008            StaticFileSegment::Headers | StaticFileSegment::Transactions => None,
2009        }
2010    }
2011
2012    /// Returns the earliest available block number that has not been expired and is still
2013    /// available.
2014    ///
2015    /// This means that the highest expired block (or expired block height) is
2016    /// `earliest_history_height.saturating_sub(1)`.
2017    ///
2018    /// Returns `0` if no history has been expired.
2019    pub fn earliest_history_height(&self) -> BlockNumber {
2020        self.earliest_history_height.load(std::sync::atomic::Ordering::Relaxed)
2021    }
2022
2023    /// Sets the earliest available block number, matching what [`Self::initialize_index`] derives.
2024    pub(crate) fn set_earliest_history_height(&self, block: BlockNumber) {
2025        self.earliest_history_height.store(block, std::sync::atomic::Ordering::Relaxed);
2026    }
2027
2028    /// Gets the lowest static file's block range if it exists for a static file segment.
2029    ///
2030    /// If there is nothing on disk for the given segment, this will return [`None`].
2031    pub fn get_lowest_range(&self, segment: StaticFileSegment) -> Option<SegmentRangeInclusive> {
2032        self.indexes.read().get(segment).and_then(|index| index.min_block_range)
2033    }
2034
2035    /// Gets the lowest static file's block range start if it exists for a static file segment.
2036    ///
2037    /// For example if the lowest static file has blocks 0-499, this will return 0.
2038    ///
2039    /// If there is nothing on disk for the given segment, this will return [`None`].
2040    pub fn get_lowest_range_start(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
2041        self.get_lowest_range(segment).map(|range| range.start())
2042    }
2043
2044    /// Gets the lowest static file's block range end if it exists for a static file segment.
2045    ///
2046    /// For example if the static file has blocks 0-499, this will return 499.
2047    ///
2048    /// If there is nothing on disk for the given segment, this will return [`None`].
2049    pub fn get_lowest_range_end(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
2050        self.get_lowest_range(segment).map(|range| range.end())
2051    }
2052
2053    /// Gets the highest static file's block height if it exists for a static file segment.
2054    ///
2055    /// If there is nothing on disk for the given segment, this will return [`None`].
2056    pub fn get_highest_static_file_block(&self, segment: StaticFileSegment) -> Option<BlockNumber> {
2057        self.indexes.read().get(segment).map(|index| index.max_block)
2058    }
2059
2060    /// Converts a range to a bounded `RangeInclusive` capped to the highest static file block.
2061    ///
2062    /// This is necessary because static file iteration beyond the tip would loop forever:
2063    /// blocks beyond the static file tip return `Ok(empty)` which is indistinguishable from
2064    /// blocks with no changes. We cap the end to the highest available block regardless of
2065    /// whether the input was unbounded or an explicit large value like `BlockNumber::MAX`.
2066    fn bound_range(
2067        &self,
2068        range: impl RangeBounds<BlockNumber>,
2069        segment: StaticFileSegment,
2070    ) -> RangeInclusive<BlockNumber> {
2071        let highest_block = self.get_highest_static_file_block(segment).unwrap_or(0);
2072
2073        let start = match range.start_bound() {
2074            Bound::Included(&n) => n,
2075            Bound::Excluded(&n) => n.saturating_add(1),
2076            Bound::Unbounded => 0,
2077        };
2078        let end = match range.end_bound() {
2079            Bound::Included(&n) => n.min(highest_block),
2080            Bound::Excluded(&n) => n.saturating_sub(1).min(highest_block),
2081            Bound::Unbounded => highest_block,
2082        };
2083
2084        start..=end
2085    }
2086
2087    /// Gets the highest static file transaction.
2088    ///
2089    /// If there is nothing on disk for the given segment, this will return [`None`].
2090    pub fn get_highest_static_file_tx(&self, segment: StaticFileSegment) -> Option<TxNumber> {
2091        self.indexes
2092            .read()
2093            .get(segment)
2094            .and_then(|index| index.available_block_ranges_by_max_tx.as_ref())
2095            .and_then(|index| index.last_key_value().map(|(last_tx, _)| *last_tx))
2096    }
2097
2098    /// Gets the highest static file block for all segments.
2099    pub fn get_highest_static_files(&self) -> HighestStaticFiles {
2100        HighestStaticFiles {
2101            receipts: self.get_highest_static_file_block(StaticFileSegment::Receipts),
2102        }
2103    }
2104
2105    /// Iterates through segment `static_files` in reverse order, executing a function until it
2106    /// returns some object. Useful for finding objects by [`TxHash`] or [`BlockHash`].
2107    pub fn find_static_file<T>(
2108        &self,
2109        segment: StaticFileSegment,
2110        func: impl Fn(StaticFileJarProvider<'_, N>) -> ProviderResult<Option<T>>,
2111    ) -> ProviderResult<Option<T>> {
2112        if let Some(ranges) =
2113            self.indexes.read().get(segment).map(|index| &index.expected_block_ranges_by_max_block)
2114        {
2115            // Iterate through all ranges in reverse order (highest to lowest)
2116            for range in ranges.values().rev() {
2117                if let Some(res) = func(self.get_or_create_jar_provider(segment, range)?)? {
2118                    return Ok(Some(res));
2119                }
2120            }
2121        }
2122
2123        Ok(None)
2124    }
2125
2126    /// Fetches data within a specified range across multiple static files.
2127    ///
2128    /// This function iteratively retrieves data using `get_fn` for each item in the given range.
2129    /// It continues fetching until the end of the range is reached or the provided `predicate`
2130    /// returns false.
2131    pub fn fetch_range_with_predicate<T, F, P>(
2132        &self,
2133        segment: StaticFileSegment,
2134        range: Range<u64>,
2135        mut get_fn: F,
2136        mut predicate: P,
2137    ) -> ProviderResult<Vec<T>>
2138    where
2139        F: FnMut(&mut StaticFileCursor<'_>, u64) -> ProviderResult<Option<T>>,
2140        P: FnMut(&T) -> bool,
2141    {
2142        let mut result = Vec::with_capacity((range.end - range.start).min(100) as usize);
2143
2144        /// Resolves to the provider for the given block or transaction number.
2145        ///
2146        /// If the static file is missing, the `result` is returned.
2147        macro_rules! get_provider {
2148            ($number:expr) => {{
2149                match self.get_segment_provider(segment, $number) {
2150                    Ok(provider) => provider,
2151                    Err(
2152                        ProviderError::MissingStaticFileBlock(_, _) |
2153                        ProviderError::MissingStaticFileTx(_, _),
2154                    ) => return Ok(result),
2155                    Err(err) => return Err(err),
2156                }
2157            }};
2158        }
2159
2160        let mut provider = get_provider!(range.start);
2161        let mut cursor = provider.cursor()?;
2162
2163        // advances number in range
2164        'outer: for number in range {
2165            // The `retrying` flag ensures a single retry attempt per `number`. If `get_fn` fails to
2166            // access data in two different static files, it halts further attempts by returning
2167            // an error, effectively preventing infinite retry loops.
2168            let mut retrying = false;
2169
2170            // advances static files if `get_fn` returns None
2171            'inner: loop {
2172                match get_fn(&mut cursor, number)? {
2173                    Some(res) => {
2174                        if !predicate(&res) {
2175                            break 'outer;
2176                        }
2177                        result.push(res);
2178                        break 'inner;
2179                    }
2180                    None => {
2181                        if retrying {
2182                            return Ok(result);
2183                        }
2184                        // There is a very small chance of hitting a deadlock if two consecutive
2185                        // static files share the same bucket in the
2186                        // internal dashmap and we don't drop the current provider
2187                        // before requesting the next one.
2188                        drop(cursor);
2189                        drop(provider);
2190                        provider = get_provider!(number);
2191                        cursor = provider.cursor()?;
2192                        retrying = true;
2193                    }
2194                }
2195            }
2196        }
2197
2198        result.shrink_to_fit();
2199
2200        Ok(result)
2201    }
2202
2203    /// Fetches data within a specified range across multiple static files.
2204    ///
2205    /// Returns an iterator over the data. Yields [`None`] if the data for the specified number is
2206    /// not found.
2207    pub fn fetch_range_iter<'a, T, F>(
2208        &'a self,
2209        segment: StaticFileSegment,
2210        range: Range<u64>,
2211        get_fn: F,
2212    ) -> ProviderResult<impl Iterator<Item = ProviderResult<Option<T>>> + 'a>
2213    where
2214        F: Fn(&mut StaticFileCursor<'_>, u64) -> ProviderResult<Option<T>> + 'a,
2215        T: std::fmt::Debug,
2216    {
2217        let mut provider = self.get_maybe_segment_provider(segment, range.start)?;
2218        Ok(range.map(move |number| {
2219            match provider
2220                .as_ref()
2221                .map(|provider| get_fn(&mut provider.cursor()?, number))
2222                .and_then(|result| result.transpose())
2223            {
2224                Some(result) => result.map(Some),
2225                None => {
2226                    // There is a very small chance of hitting a deadlock if two consecutive
2227                    // static files share the same bucket in the internal dashmap and we don't drop
2228                    // the current provider before requesting the next one.
2229                    provider.take();
2230                    provider = self.get_maybe_segment_provider(segment, number)?;
2231                    provider
2232                        .as_ref()
2233                        .map(|provider| get_fn(&mut provider.cursor()?, number))
2234                        .and_then(|result| result.transpose())
2235                        .transpose()
2236                }
2237            }
2238        }))
2239    }
2240
2241    /// Returns directory where `static_files` are located.
2242    pub fn directory(&self) -> &Path {
2243        &self.path
2244    }
2245
2246    /// Retrieves data from the database or static file, wherever it's available.
2247    ///
2248    /// # Arguments
2249    /// * `segment` - The segment of the static file to check against.
2250    /// * `index_key` - Requested index key, usually a block or transaction number.
2251    /// * `fetch_from_static_file` - A closure that defines how to fetch the data from the static
2252    ///   file provider.
2253    /// * `fetch_from_database` - A closure that defines how to fetch the data from the database
2254    ///   when the static file doesn't contain the required data or is not available.
2255    pub fn get_with_static_file_or_database<T, FS, FD>(
2256        &self,
2257        segment: StaticFileSegment,
2258        number: u64,
2259        fetch_from_static_file: FS,
2260        fetch_from_database: FD,
2261    ) -> ProviderResult<Option<T>>
2262    where
2263        FS: Fn(&Self) -> ProviderResult<Option<T>>,
2264        FD: Fn() -> ProviderResult<Option<T>>,
2265    {
2266        // If there is, check the maximum block or transaction number of the segment.
2267        let static_file_upper_bound = if segment.is_block_or_change_based() {
2268            self.get_highest_static_file_block(segment)
2269        } else {
2270            self.get_highest_static_file_tx(segment)
2271        };
2272
2273        if static_file_upper_bound
2274            .is_some_and(|static_file_upper_bound| static_file_upper_bound >= number)
2275        {
2276            return fetch_from_static_file(self);
2277        }
2278        fetch_from_database()
2279    }
2280
2281    /// Gets data within a specified range, potentially spanning different `static_files` and
2282    /// database.
2283    ///
2284    /// # Arguments
2285    /// * `segment` - The segment of the static file to query.
2286    /// * `block_or_tx_range` - The range of data to fetch.
2287    /// * `fetch_from_static_file` - A function to fetch data from the `static_file`.
2288    /// * `fetch_from_database` - A function to fetch data from the database.
2289    /// * `predicate` - A function used to evaluate each item in the fetched data. Fetching is
2290    ///   terminated when this function returns false, thereby filtering the data based on the
2291    ///   provided condition.
2292    pub fn get_range_with_static_file_or_database<T, P, FS, FD>(
2293        &self,
2294        segment: StaticFileSegment,
2295        mut block_or_tx_range: Range<u64>,
2296        fetch_from_static_file: FS,
2297        mut fetch_from_database: FD,
2298        mut predicate: P,
2299    ) -> ProviderResult<Vec<T>>
2300    where
2301        FS: Fn(&Self, Range<u64>, &mut P) -> ProviderResult<Vec<T>>,
2302        FD: FnMut(Range<u64>, P) -> ProviderResult<Vec<T>>,
2303        P: FnMut(&T) -> bool,
2304    {
2305        let mut data = Vec::new();
2306
2307        // If there is, check the maximum block or transaction number of the segment.
2308        if let Some(static_file_upper_bound) = if segment.is_block_or_change_based() {
2309            self.get_highest_static_file_block(segment)
2310        } else {
2311            self.get_highest_static_file_tx(segment)
2312        } && block_or_tx_range.start <= static_file_upper_bound
2313        {
2314            let end = block_or_tx_range.end.min(static_file_upper_bound + 1);
2315            data.extend(fetch_from_static_file(
2316                self,
2317                block_or_tx_range.start..end,
2318                &mut predicate,
2319            )?);
2320            block_or_tx_range.start = end;
2321        }
2322
2323        if block_or_tx_range.end > block_or_tx_range.start {
2324            data.extend(fetch_from_database(block_or_tx_range, predicate)?)
2325        }
2326
2327        Ok(data)
2328    }
2329
2330    /// Returns static files directory
2331    #[cfg(any(test, feature = "test-utils"))]
2332    pub fn path(&self) -> &Path {
2333        &self.path
2334    }
2335
2336    /// Returns transaction index
2337    #[cfg(any(test, feature = "test-utils"))]
2338    pub fn tx_index(&self, segment: StaticFileSegment) -> Option<SegmentRanges> {
2339        self.indexes
2340            .read()
2341            .get(segment)
2342            .and_then(|index| index.available_block_ranges_by_max_tx.as_ref())
2343            .cloned()
2344    }
2345
2346    /// Returns expected block index
2347    #[cfg(any(test, feature = "test-utils"))]
2348    pub fn expected_block_index(&self, segment: StaticFileSegment) -> Option<SegmentRanges> {
2349        self.indexes
2350            .read()
2351            .get(segment)
2352            .map(|index| &index.expected_block_ranges_by_max_block)
2353            .cloned()
2354    }
2355}
2356
2357#[derive(Debug)]
2358struct StaticFileSegmentIndex {
2359    /// Min static file block range.
2360    ///
2361    /// This index is initialized on launch to keep track of the lowest, non-expired static file
2362    /// per segment and gets updated on [`StaticFileProvider::update_index`].
2363    ///
2364    /// This tracks the lowest static file per segment together with the block range in that
2365    /// file. E.g. static file is batched in 500k block intervals then the lowest static file
2366    /// is [0..499K], and the block range is start = 0, end = 499K.
2367    ///
2368    /// This index is mainly used for history expiry, which targets transactions, e.g. pre-merge
2369    /// history expiry would lead to removing all static files below the merge height.
2370    min_block_range: Option<SegmentRangeInclusive>,
2371    /// Max static file block.
2372    max_block: u64,
2373    /// Expected static file block ranges indexed by max expected blocks.
2374    ///
2375    /// For example, a static file for expected block range `0..=499_000` may have only block range
2376    /// `0..=1000` contained in it, as it's not fully filled yet. This index maps the max expected
2377    /// block to the expected range, i.e. block `499_000` to block range `0..=499_000`.
2378    expected_block_ranges_by_max_block: SegmentRanges,
2379    /// Available on disk static file block ranges indexed by max transactions.
2380    ///
2381    /// For example, a static file for block range `0..=499_000` may only have block range
2382    /// `0..=1000` and transaction range `0..=2000` contained in it. This index maps the max
2383    /// available transaction to the available block range, i.e. transaction `2000` to block range
2384    /// `0..=1000`.
2385    available_block_ranges_by_max_tx: Option<SegmentRanges>,
2386}
2387
2388/// Helper trait to manage different [`StaticFileProviderRW`] of an `Arc<StaticFileProvider`
2389pub trait StaticFileWriter {
2390    /// The primitives type used by the static file provider.
2391    type Primitives: Send + Sync + 'static;
2392
2393    /// Returns a mutable reference to a [`StaticFileProviderRW`] of a [`StaticFileSegment`].
2394    fn get_writer(
2395        &self,
2396        block: BlockNumber,
2397        segment: StaticFileSegment,
2398    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2399
2400    /// Returns a mutable reference to a [`StaticFileProviderRW`] of the latest
2401    /// [`StaticFileSegment`].
2402    fn latest_writer(
2403        &self,
2404        segment: StaticFileSegment,
2405    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>>;
2406
2407    /// Commits all changes of all [`StaticFileProviderRW`] of all [`StaticFileSegment`].
2408    fn commit(&self) -> ProviderResult<()>;
2409
2410    /// Returns `true` if the static file provider has unwind queued.
2411    fn has_unwind_queued(&self) -> bool;
2412
2413    /// Finalizes all static file writers by committing their configuration to disk.
2414    ///
2415    /// Returns an error if prune is queued (use [`Self::commit`] instead).
2416    fn finalize(&self) -> ProviderResult<()>;
2417}
2418
2419impl<N: NodePrimitives> StaticFileWriter for StaticFileProvider<N> {
2420    type Primitives = N;
2421
2422    fn get_writer(
2423        &self,
2424        block: BlockNumber,
2425        segment: StaticFileSegment,
2426    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
2427        if self.access.is_read_only() {
2428            return Err(ProviderError::ReadOnlyStaticFileAccess);
2429        }
2430
2431        trace!(target: "providers::static_file", ?block, ?segment, "Getting static file writer.");
2432        self.writers.get_or_create(segment, || {
2433            StaticFileProviderRW::new(segment, block, Arc::downgrade(&self.0), self.metrics.clone())
2434        })
2435    }
2436
2437    fn latest_writer(
2438        &self,
2439        segment: StaticFileSegment,
2440    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
2441        let genesis_number = self.0.as_ref().genesis_block_number();
2442        self.get_writer(
2443            self.get_highest_static_file_block(segment).unwrap_or(genesis_number),
2444            segment,
2445        )
2446    }
2447
2448    fn commit(&self) -> ProviderResult<()> {
2449        self.writers.commit()
2450    }
2451
2452    fn has_unwind_queued(&self) -> bool {
2453        self.writers.has_unwind_queued()
2454    }
2455
2456    fn finalize(&self) -> ProviderResult<()> {
2457        self.writers.finalize()
2458    }
2459}
2460
2461impl<N: NodePrimitives> ChangeSetReader for StaticFileProvider<N> {
2462    fn account_block_changeset(
2463        &self,
2464        block_number: BlockNumber,
2465    ) -> ProviderResult<Vec<reth_db::models::AccountBeforeTx>> {
2466        let provider = match self.get_segment_provider_for_block(
2467            StaticFileSegment::AccountChangeSets,
2468            block_number,
2469            None,
2470        ) {
2471            Ok(provider) => provider,
2472            Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(Vec::new()),
2473            Err(err) => return Err(err),
2474        };
2475
2476        if let Some(offset) = provider.read_changeset_offset(block_number)? {
2477            let mut cursor = provider.cursor()?;
2478            let mut changeset = Vec::with_capacity(offset.num_changes() as usize);
2479
2480            for i in offset.changeset_range() {
2481                if let Some(change) =
2482                    cursor.get_one::<reth_db::static_file::AccountChangesetMask>(i.into())?
2483                {
2484                    changeset.push(change)
2485                }
2486            }
2487            Ok(changeset)
2488        } else {
2489            Ok(Vec::new())
2490        }
2491    }
2492
2493    fn get_account_before_block(
2494        &self,
2495        block_number: BlockNumber,
2496        address: Address,
2497    ) -> ProviderResult<Option<reth_db::models::AccountBeforeTx>> {
2498        let provider = match self.get_segment_provider_for_block(
2499            StaticFileSegment::AccountChangeSets,
2500            block_number,
2501            None,
2502        ) {
2503            Ok(provider) => provider,
2504            Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(None),
2505            Err(err) => return Err(err),
2506        };
2507
2508        let Some(offset) = provider.read_changeset_offset(block_number)? else {
2509            return Ok(None);
2510        };
2511
2512        let mut cursor = provider.cursor()?;
2513        let range = offset.changeset_range();
2514        let mut low = range.start;
2515        let mut high = range.end;
2516
2517        while low < high {
2518            let mid = low + (high - low) / 2;
2519            if let Some(change) =
2520                cursor.get_one::<reth_db::static_file::AccountChangesetMask>(mid.into())?
2521            {
2522                if change.address < address {
2523                    low = mid + 1;
2524                } else {
2525                    high = mid;
2526                }
2527            } else {
2528                // This is not expected but means we are out of the range / file somehow, and can't
2529                // continue
2530                debug!(
2531                    target: "providers::static_file",
2532                    ?low,
2533                    ?mid,
2534                    ?high,
2535                    ?range,
2536                    ?block_number,
2537                    ?address,
2538                    "Cannot continue binary search for account changeset fetch"
2539                );
2540                low = range.end;
2541                break;
2542            }
2543        }
2544
2545        if low < range.end &&
2546            let Some(change) = cursor
2547                .get_one::<reth_db::static_file::AccountChangesetMask>(low.into())?
2548                .filter(|change| change.address == address)
2549        {
2550            return Ok(Some(change));
2551        }
2552
2553        Ok(None)
2554    }
2555
2556    fn account_changesets_range(
2557        &self,
2558        range: impl core::ops::RangeBounds<BlockNumber>,
2559    ) -> ProviderResult<Vec<(BlockNumber, reth_db::models::AccountBeforeTx)>> {
2560        let range = self.bound_range(range, StaticFileSegment::AccountChangeSets);
2561        self.walk_account_changeset_range(range).collect()
2562    }
2563}
2564
2565impl<N: NodePrimitives> StorageChangeSetReader for StaticFileProvider<N> {
2566    fn storage_changeset(
2567        &self,
2568        block_number: BlockNumber,
2569    ) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
2570        let provider = match self.get_segment_provider_for_block(
2571            StaticFileSegment::StorageChangeSets,
2572            block_number,
2573            None,
2574        ) {
2575            Ok(provider) => provider,
2576            Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(Vec::new()),
2577            Err(err) => return Err(err),
2578        };
2579
2580        if let Some(offset) = provider.read_changeset_offset(block_number)? {
2581            let mut cursor = provider.cursor()?;
2582            let mut changeset = Vec::with_capacity(offset.num_changes() as usize);
2583
2584            for i in offset.changeset_range() {
2585                if let Some(change) = cursor.get_one::<StorageChangesetMask>(i.into())? {
2586                    let block_address = BlockNumberAddress((block_number, change.address));
2587                    let entry = StorageEntry { key: change.key, value: change.value };
2588                    changeset.push((block_address, entry));
2589                }
2590            }
2591            Ok(changeset)
2592        } else {
2593            Ok(Vec::new())
2594        }
2595    }
2596
2597    fn get_storage_before_block(
2598        &self,
2599        block_number: BlockNumber,
2600        address: Address,
2601        storage_key: B256,
2602    ) -> ProviderResult<Option<StorageEntry>> {
2603        let provider = match self.get_segment_provider_for_block(
2604            StaticFileSegment::StorageChangeSets,
2605            block_number,
2606            None,
2607        ) {
2608            Ok(provider) => provider,
2609            Err(ProviderError::MissingStaticFileBlock(_, _)) => return Ok(None),
2610            Err(err) => return Err(err),
2611        };
2612
2613        let Some(offset) = provider.read_changeset_offset(block_number)? else {
2614            return Ok(None);
2615        };
2616
2617        let mut cursor = provider.cursor()?;
2618        let range = offset.changeset_range();
2619        let mut low = range.start;
2620        let mut high = range.end;
2621
2622        while low < high {
2623            let mid = low + (high - low) / 2;
2624            if let Some(change) = cursor.get_one::<StorageChangesetMask>(mid.into())? {
2625                match (change.address, change.key).cmp(&(address, storage_key)) {
2626                    std::cmp::Ordering::Less => low = mid + 1,
2627                    _ => high = mid,
2628                }
2629            } else {
2630                debug!(
2631                    target: "providers::static_file",
2632                    ?low,
2633                    ?mid,
2634                    ?high,
2635                    ?range,
2636                    ?block_number,
2637                    ?address,
2638                    ?storage_key,
2639                    "Cannot continue binary search for storage changeset fetch"
2640                );
2641                low = range.end;
2642                break;
2643            }
2644        }
2645
2646        if low < range.end &&
2647            let Some(change) = cursor
2648                .get_one::<StorageChangesetMask>(low.into())?
2649                .filter(|change| change.address == address && change.key == storage_key)
2650        {
2651            return Ok(Some(StorageEntry { key: change.key, value: change.value }));
2652        }
2653
2654        Ok(None)
2655    }
2656
2657    fn storage_changesets_range(
2658        &self,
2659        range: impl RangeBounds<BlockNumber>,
2660    ) -> ProviderResult<Vec<(BlockNumberAddress, StorageEntry)>> {
2661        let range = self.bound_range(range, StaticFileSegment::StorageChangeSets);
2662        self.walk_storage_changeset_range(range).collect()
2663    }
2664}
2665
2666impl<N: NodePrimitives> StaticFileProvider<N> {
2667    /// Creates an iterator for walking through account changesets in the specified block range.
2668    ///
2669    /// This returns a lazy iterator that fetches changesets block by block to avoid loading
2670    /// everything into memory at once.
2671    ///
2672    /// Accepts any range type that implements `RangeBounds<BlockNumber>`, including:
2673    /// - `Range<BlockNumber>` (e.g., `0..100`)
2674    /// - `RangeInclusive<BlockNumber>` (e.g., `0..=99`)
2675    /// - `RangeFrom<BlockNumber>` (e.g., `0..`) - iterates until exhausted
2676    pub fn walk_account_changeset_range(
2677        &self,
2678        range: impl RangeBounds<BlockNumber>,
2679    ) -> StaticFileAccountChangesetWalker<Self> {
2680        StaticFileAccountChangesetWalker::new(self.clone(), range)
2681    }
2682
2683    /// Creates an iterator for walking through storage changesets in the specified block range.
2684    pub fn walk_storage_changeset_range(
2685        &self,
2686        range: impl RangeBounds<BlockNumber>,
2687    ) -> StaticFileStorageChangesetWalker<Self> {
2688        StaticFileStorageChangesetWalker::new(self.clone(), range)
2689    }
2690}
2691
2692impl<N: NodePrimitives<BlockHeader: Value>> HeaderProvider for StaticFileProvider<N> {
2693    type Header = N::BlockHeader;
2694
2695    fn header(&self, block_hash: BlockHash) -> ProviderResult<Option<Self::Header>> {
2696        self.find_static_file(StaticFileSegment::Headers, |jar_provider| {
2697            Ok(jar_provider
2698                .cursor()?
2699                .get_two::<HeaderWithHashMask<Self::Header>>((&block_hash).into())?
2700                .and_then(|(header, hash)| {
2701                    if hash == block_hash {
2702                        return Some(header);
2703                    }
2704                    None
2705                }))
2706        })
2707    }
2708
2709    fn header_by_number(&self, num: BlockNumber) -> ProviderResult<Option<Self::Header>> {
2710        self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2711            .and_then(|provider| provider.header_by_number(num))
2712            .or_else(|err| {
2713                if let ProviderError::MissingStaticFileBlock(_, _) = err {
2714                    Ok(None)
2715                } else {
2716                    Err(err)
2717                }
2718            })
2719    }
2720
2721    fn headers_range(
2722        &self,
2723        range: impl RangeBounds<BlockNumber>,
2724    ) -> ProviderResult<Vec<Self::Header>> {
2725        self.fetch_range_with_predicate(
2726            StaticFileSegment::Headers,
2727            to_range(range),
2728            |cursor, number| cursor.get_one::<HeaderMask<Self::Header>>(number.into()),
2729            |_| true,
2730        )
2731    }
2732
2733    fn sealed_header(
2734        &self,
2735        num: BlockNumber,
2736    ) -> ProviderResult<Option<SealedHeader<Self::Header>>> {
2737        self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2738            .and_then(|provider| provider.sealed_header(num))
2739            .or_else(|err| {
2740                if let ProviderError::MissingStaticFileBlock(_, _) = err {
2741                    Ok(None)
2742                } else {
2743                    Err(err)
2744                }
2745            })
2746    }
2747
2748    fn sealed_headers_while(
2749        &self,
2750        range: impl RangeBounds<BlockNumber>,
2751        predicate: impl FnMut(&SealedHeader<Self::Header>) -> bool,
2752    ) -> ProviderResult<Vec<SealedHeader<Self::Header>>> {
2753        self.fetch_range_with_predicate(
2754            StaticFileSegment::Headers,
2755            to_range(range),
2756            |cursor, number| {
2757                Ok(cursor
2758                    .get_two::<HeaderWithHashMask<Self::Header>>(number.into())?
2759                    .map(|(header, hash)| SealedHeader::new(header, hash)))
2760            },
2761            predicate,
2762        )
2763    }
2764}
2765
2766impl<N: NodePrimitives> BlockHashReader for StaticFileProvider<N> {
2767    fn block_hash(&self, num: u64) -> ProviderResult<Option<B256>> {
2768        self.get_segment_provider_for_block(StaticFileSegment::Headers, num, None)
2769            .and_then(|provider| provider.block_hash(num))
2770            .or_else(|err| {
2771                if let ProviderError::MissingStaticFileBlock(_, _) = err {
2772                    Ok(None)
2773                } else {
2774                    Err(err)
2775                }
2776            })
2777    }
2778
2779    fn canonical_hashes_range(
2780        &self,
2781        start: BlockNumber,
2782        end: BlockNumber,
2783    ) -> ProviderResult<Vec<B256>> {
2784        self.fetch_range_with_predicate(
2785            StaticFileSegment::Headers,
2786            start..end,
2787            |cursor, number| cursor.get_one::<BlockHashMask>(number.into()),
2788            |_| true,
2789        )
2790    }
2791}
2792
2793impl<N: NodePrimitives<SignedTx: Value + SignedTransaction, Receipt: Value>> ReceiptProvider
2794    for StaticFileProvider<N>
2795{
2796    type Receipt = N::Receipt;
2797
2798    fn receipt(&self, num: TxNumber) -> ProviderResult<Option<Self::Receipt>> {
2799        self.get_segment_provider_for_transaction(StaticFileSegment::Receipts, num, None)
2800            .and_then(|provider| provider.receipt(num))
2801            .or_else(|err| {
2802                if let ProviderError::MissingStaticFileTx(_, _) = err {
2803                    Ok(None)
2804                } else {
2805                    Err(err)
2806                }
2807            })
2808    }
2809
2810    fn receipt_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Receipt>> {
2811        if let Some(num) = self.transaction_id(hash)? {
2812            return self.receipt(num);
2813        }
2814        Ok(None)
2815    }
2816
2817    fn receipts_by_block(
2818        &self,
2819        _block: BlockHashOrNumber,
2820    ) -> ProviderResult<Option<Vec<Self::Receipt>>> {
2821        unreachable!()
2822    }
2823
2824    fn receipts_by_tx_range(
2825        &self,
2826        range: impl RangeBounds<TxNumber>,
2827    ) -> ProviderResult<Vec<Self::Receipt>> {
2828        self.fetch_range_with_predicate(
2829            StaticFileSegment::Receipts,
2830            to_range(range),
2831            |cursor, number| cursor.get_one::<ReceiptMask<Self::Receipt>>(number.into()),
2832            |_| true,
2833        )
2834    }
2835
2836    fn receipts_by_block_range(
2837        &self,
2838        _block_range: RangeInclusive<BlockNumber>,
2839    ) -> ProviderResult<Vec<Vec<Self::Receipt>>> {
2840        Err(ProviderError::UnsupportedProvider)
2841    }
2842}
2843
2844impl<N: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>> TransactionsProviderExt
2845    for StaticFileProvider<N>
2846{
2847    fn transaction_hashes_by_range(
2848        &self,
2849        tx_range: Range<TxNumber>,
2850    ) -> ProviderResult<Vec<(TxHash, TxNumber)>> {
2851        let tx_range_size = (tx_range.end - tx_range.start) as usize;
2852
2853        // Transactions are different size, so chunks will not all take the same processing time. If
2854        // chunks are too big, there will be idle threads waiting for work. Choosing an
2855        // arbitrary smaller value to make sure it doesn't happen.
2856        let chunk_size = 100;
2857
2858        // iterator over the chunks
2859        let chunks = tx_range
2860            .clone()
2861            .step_by(chunk_size)
2862            .map(|start| start..std::cmp::min(start + chunk_size as u64, tx_range.end));
2863        let mut channels = Vec::with_capacity(tx_range_size.div_ceil(chunk_size));
2864
2865        for chunk_range in chunks {
2866            let (channel_tx, channel_rx) = mpsc::channel();
2867            channels.push(channel_rx);
2868
2869            let manager = self.clone();
2870
2871            // Spawn the task onto the global rayon pool
2872            // This task will send the cached transaction hash through the channel.
2873            rayon::spawn(move || {
2874                let _ = manager.fetch_range_with_predicate(
2875                    StaticFileSegment::Transactions,
2876                    chunk_range,
2877                    |cursor, number| {
2878                        Ok(cursor
2879                            .get_one::<TransactionMask<Self::Transaction>>(number.into())?
2880                            .map(|transaction| {
2881                                let _ = channel_tx.send(transaction_hash((number, transaction)));
2882                            }))
2883                    },
2884                    |_| true,
2885                );
2886            });
2887        }
2888
2889        let mut tx_list = Vec::with_capacity(tx_range_size);
2890
2891        // Iterate over channels and append the tx hashes unsorted
2892        for channel in channels {
2893            while let Ok(tx) = channel.recv() {
2894                let (tx_hash, tx_id) = tx.map_err(|boxed| *boxed)?;
2895                tx_list.push((tx_hash, tx_id));
2896            }
2897        }
2898
2899        Ok(tx_list)
2900    }
2901}
2902
2903impl<N: NodePrimitives<SignedTx: Decompress + SignedTransaction>> TransactionsProvider
2904    for StaticFileProvider<N>
2905{
2906    type Transaction = N::SignedTx;
2907
2908    fn transaction_id(&self, tx_hash: TxHash) -> ProviderResult<Option<TxNumber>> {
2909        self.find_static_file(StaticFileSegment::Transactions, |jar_provider| {
2910            let mut cursor = jar_provider.cursor()?;
2911            if cursor
2912                .get_one::<TransactionMask<Self::Transaction>>((&tx_hash).into())?
2913                .and_then(|tx| (*tx.tx_hash() == tx_hash).then_some(tx))
2914                .is_some()
2915            {
2916                Ok(cursor.number())
2917            } else {
2918                Ok(None)
2919            }
2920        })
2921    }
2922
2923    fn transaction_by_id(&self, num: TxNumber) -> ProviderResult<Option<Self::Transaction>> {
2924        self.get_segment_provider_for_transaction(StaticFileSegment::Transactions, num, None)
2925            .and_then(|provider| provider.transaction_by_id(num))
2926            .or_else(|err| {
2927                if let ProviderError::MissingStaticFileTx(_, _) = err {
2928                    Ok(None)
2929                } else {
2930                    Err(err)
2931                }
2932            })
2933    }
2934
2935    fn transaction_by_id_unhashed(
2936        &self,
2937        num: TxNumber,
2938    ) -> ProviderResult<Option<Self::Transaction>> {
2939        self.get_segment_provider_for_transaction(StaticFileSegment::Transactions, num, None)
2940            .and_then(|provider| provider.transaction_by_id_unhashed(num))
2941            .or_else(|err| {
2942                if let ProviderError::MissingStaticFileTx(_, _) = err {
2943                    Ok(None)
2944                } else {
2945                    Err(err)
2946                }
2947            })
2948    }
2949
2950    fn transaction_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Transaction>> {
2951        self.find_static_file(StaticFileSegment::Transactions, |jar_provider| {
2952            Ok(jar_provider
2953                .cursor()?
2954                .get_one::<TransactionMask<Self::Transaction>>((&hash).into())?
2955                .and_then(|tx| (*tx.tx_hash() == hash).then_some(tx)))
2956        })
2957    }
2958
2959    fn transaction_by_hash_with_meta(
2960        &self,
2961        _hash: TxHash,
2962    ) -> ProviderResult<Option<(Self::Transaction, TransactionMeta)>> {
2963        // Required data not present in static_files
2964        Err(ProviderError::UnsupportedProvider)
2965    }
2966
2967    fn transactions_by_block(
2968        &self,
2969        _block_id: BlockHashOrNumber,
2970    ) -> ProviderResult<Option<Vec<Self::Transaction>>> {
2971        // Required data not present in static_files
2972        Err(ProviderError::UnsupportedProvider)
2973    }
2974
2975    fn transactions_by_block_range(
2976        &self,
2977        _range: impl RangeBounds<BlockNumber>,
2978    ) -> ProviderResult<Vec<Vec<Self::Transaction>>> {
2979        // Required data not present in static_files
2980        Err(ProviderError::UnsupportedProvider)
2981    }
2982
2983    fn transactions_by_tx_range(
2984        &self,
2985        range: impl RangeBounds<TxNumber>,
2986    ) -> ProviderResult<Vec<Self::Transaction>> {
2987        self.fetch_range_with_predicate(
2988            StaticFileSegment::Transactions,
2989            to_range(range),
2990            |cursor, number| cursor.get_one::<TransactionMask<Self::Transaction>>(number.into()),
2991            |_| true,
2992        )
2993    }
2994
2995    fn senders_by_tx_range(
2996        &self,
2997        range: impl RangeBounds<TxNumber>,
2998    ) -> ProviderResult<Vec<Address>> {
2999        self.fetch_range_with_predicate(
3000            StaticFileSegment::TransactionSenders,
3001            to_range(range),
3002            |cursor, number| cursor.get_one::<TransactionSenderMask>(number.into()),
3003            |_| true,
3004        )
3005    }
3006
3007    fn transaction_sender(&self, id: TxNumber) -> ProviderResult<Option<Address>> {
3008        self.get_segment_provider_for_transaction(StaticFileSegment::TransactionSenders, id, None)
3009            .and_then(|provider| provider.transaction_sender(id))
3010            .or_else(|err| {
3011                if let ProviderError::MissingStaticFileTx(_, _) = err {
3012                    Ok(None)
3013                } else {
3014                    Err(err)
3015                }
3016            })
3017    }
3018}
3019
3020impl<N: NodePrimitives> BlockNumReader for StaticFileProvider<N> {
3021    fn chain_info(&self) -> ProviderResult<ChainInfo> {
3022        // Required data not present in static_files
3023        Err(ProviderError::UnsupportedProvider)
3024    }
3025
3026    fn best_block_number(&self) -> ProviderResult<BlockNumber> {
3027        // Required data not present in static_files
3028        Err(ProviderError::UnsupportedProvider)
3029    }
3030
3031    fn last_block_number(&self) -> ProviderResult<BlockNumber> {
3032        Ok(self.get_highest_static_file_block(StaticFileSegment::Headers).unwrap_or_default())
3033    }
3034
3035    fn block_number(&self, _hash: B256) -> ProviderResult<Option<BlockNumber>> {
3036        // Required data not present in static_files
3037        Err(ProviderError::UnsupportedProvider)
3038    }
3039}
3040
3041/* Cannot be successfully implemented but must exist for trait requirements */
3042
3043impl<N: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>> BlockReader
3044    for StaticFileProvider<N>
3045{
3046    type Block = N::Block;
3047
3048    fn find_block_by_hash(
3049        &self,
3050        _hash: B256,
3051        _source: BlockSource,
3052    ) -> ProviderResult<Option<Self::Block>> {
3053        // Required data not present in static_files
3054        Err(ProviderError::UnsupportedProvider)
3055    }
3056
3057    fn block(&self, _id: BlockHashOrNumber) -> ProviderResult<Option<Self::Block>> {
3058        // Required data not present in static_files
3059        Err(ProviderError::UnsupportedProvider)
3060    }
3061
3062    fn pending_block(&self) -> ProviderResult<Option<Arc<RecoveredBlock<Self::Block>>>> {
3063        // Required data not present in static_files
3064        Err(ProviderError::UnsupportedProvider)
3065    }
3066
3067    fn pending_block_and_receipts(
3068        &self,
3069    ) -> ProviderResult<Option<RecoveredBlockAndExecutionOutput<Self::Block, Self::Receipt>>> {
3070        // Required data not present in static_files
3071        Err(ProviderError::UnsupportedProvider)
3072    }
3073
3074    fn recovered_block(
3075        &self,
3076        _id: BlockHashOrNumber,
3077        _transaction_kind: TransactionVariant,
3078    ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
3079        // Required data not present in static_files
3080        Err(ProviderError::UnsupportedProvider)
3081    }
3082
3083    fn sealed_block_with_senders(
3084        &self,
3085        _id: BlockHashOrNumber,
3086        _transaction_kind: TransactionVariant,
3087    ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
3088        // Required data not present in static_files
3089        Err(ProviderError::UnsupportedProvider)
3090    }
3091
3092    fn block_range(&self, _range: RangeInclusive<BlockNumber>) -> ProviderResult<Vec<Self::Block>> {
3093        // Required data not present in static_files
3094        Err(ProviderError::UnsupportedProvider)
3095    }
3096
3097    fn block_with_senders_range(
3098        &self,
3099        _range: RangeInclusive<BlockNumber>,
3100    ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
3101        Err(ProviderError::UnsupportedProvider)
3102    }
3103
3104    fn recovered_block_range(
3105        &self,
3106        _range: RangeInclusive<BlockNumber>,
3107    ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
3108        Err(ProviderError::UnsupportedProvider)
3109    }
3110
3111    fn block_by_transaction_id(&self, _id: TxNumber) -> ProviderResult<Option<BlockNumber>> {
3112        Err(ProviderError::UnsupportedProvider)
3113    }
3114}
3115
3116impl<N: NodePrimitives> BlockBodyIndicesProvider for StaticFileProvider<N> {
3117    fn block_body_indices(&self, _num: u64) -> ProviderResult<Option<StoredBlockBodyIndices>> {
3118        Err(ProviderError::UnsupportedProvider)
3119    }
3120
3121    fn block_body_indices_range(
3122        &self,
3123        _range: RangeInclusive<BlockNumber>,
3124    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
3125        Err(ProviderError::UnsupportedProvider)
3126    }
3127}
3128
3129impl<N: NodePrimitives> StatsReader for StaticFileProvider<N> {
3130    fn count_entries<T: Table>(&self) -> ProviderResult<usize> {
3131        match T::NAME {
3132            tables::CanonicalHeaders::NAME |
3133            tables::Headers::<Header>::NAME |
3134            tables::HeaderTerminalDifficulties::NAME => Ok(self
3135                .get_highest_static_file_block(StaticFileSegment::Headers)
3136                .map(|block| block + 1)
3137                .unwrap_or_default()
3138                as usize),
3139            tables::Receipts::<Receipt>::NAME => Ok(self
3140                .get_highest_static_file_tx(StaticFileSegment::Receipts)
3141                .map(|receipts| receipts + 1)
3142                .unwrap_or_default() as usize),
3143            tables::Transactions::<TransactionSigned>::NAME => Ok(self
3144                .get_highest_static_file_tx(StaticFileSegment::Transactions)
3145                .map(|txs| txs + 1)
3146                .unwrap_or_default()
3147                as usize),
3148            tables::TransactionSenders::NAME => Ok(self
3149                .get_highest_static_file_tx(StaticFileSegment::TransactionSenders)
3150                .map(|txs| txs + 1)
3151                .unwrap_or_default() as usize),
3152            _ => Err(ProviderError::UnsupportedProvider),
3153        }
3154    }
3155}
3156
3157/// Returns the tx hash for the given transaction and its id.
3158#[inline]
3159fn transaction_hash<T>(entry: (TxNumber, T)) -> Result<(B256, TxNumber), Box<ProviderError>>
3160where
3161    T: TxHashRef,
3162{
3163    let (tx_id, tx) = entry;
3164    Ok((*tx.tx_hash(), tx_id))
3165}
3166
3167#[cfg(test)]
3168mod tests {
3169    use std::{
3170        collections::BTreeMap,
3171        sync::{atomic::Ordering, mpsc},
3172        thread,
3173        time::{Duration, Instant},
3174    };
3175
3176    use alloy_consensus::Header;
3177    use alloy_primitives::B256;
3178    use reth_chain_state::EthPrimitives;
3179    use reth_db::test_utils::create_test_static_files_dir;
3180    use reth_static_file_types::{SegmentRangeInclusive, StaticFileSegment};
3181
3182    use super::StaticFileWriter;
3183    use crate::{providers::StaticFileProvider, BlockHashReader, StaticFileProviderBuilder};
3184
3185    #[test]
3186    fn recovery_prunes_transaction_zero_at_empty_genesis() -> eyre::Result<()> {
3187        use crate::{
3188            test_utils::create_test_provider_factory, StaticFileProviderFactory,
3189            TransactionsProvider,
3190        };
3191        use alloy_consensus::{SignableTransaction, TxLegacy};
3192        use alloy_primitives::Signature;
3193        use reth_db::{tables, transaction::DbTxMut};
3194
3195        let factory = create_test_provider_factory();
3196        let provider = factory.provider_rw()?;
3197        provider.tx_ref().put::<tables::BlockBodyIndices>(0, Default::default())?;
3198        provider.commit()?;
3199
3200        let static_files = factory.static_file_provider();
3201        let segment = StaticFileSegment::Transactions;
3202        let tx = TxLegacy::default().into_signed(Signature::test_signature()).into();
3203        {
3204            let mut writer = static_files.latest_writer(segment)?;
3205            writer.increment_block(0)?;
3206            writer.increment_block(1)?;
3207            writer.append_transaction(0, &tx)?;
3208            writer.commit()?;
3209        }
3210        assert!(static_files.transaction_by_id(0)?.is_some());
3211
3212        assert_eq!(static_files.check_consistency(&factory.provider()?)?, None);
3213        assert_eq!(static_files.get_highest_static_file_block(segment), Some(0));
3214        assert!(static_files.transaction_by_id(0)?.is_none());
3215
3216        let mut writer = static_files.latest_writer(segment)?;
3217        writer.increment_block(1)?;
3218        writer.append_transaction(0, &tx)?;
3219        writer.commit()?;
3220        Ok(())
3221    }
3222
3223    #[test]
3224    fn stale_cache_fill_does_not_survive_index_reinitialization() -> eyre::Result<()> {
3225        let (static_dir, _) = create_test_static_files_dir();
3226        let static_files: StaticFileProvider<EthPrimitives> =
3227            StaticFileProviderBuilder::read_write(&static_dir)
3228                .with_blocks_per_file_for_segment(StaticFileSegment::Headers, 10)
3229                .with_blocks_per_file_for_segment(StaticFileSegment::Receipts, 2)
3230                .build()?;
3231
3232        let hash_0 = B256::repeat_byte(0x10);
3233        let header_0 = Header { number: 0, ..Default::default() };
3234        {
3235            let mut writer = static_files.latest_writer(StaticFileSegment::Headers)?;
3236            writer.append_header(&header_0, &hash_0)?;
3237            writer.commit()?;
3238        }
3239
3240        // Create two receipt jars. Deleting either one reinitializes every segment index and
3241        // invalidates the shared jar cache, matching the pruning path from the incidents.
3242        {
3243            let mut writer = static_files.latest_writer(StaticFileSegment::Receipts)?;
3244            for block in 0..=3 {
3245                writer.increment_block(block)?;
3246            }
3247            writer.commit()?;
3248        }
3249        static_files.delete_jar(StaticFileSegment::Receipts, 0)?;
3250
3251        let (loaded_tx, loaded_rx) = mpsc::sync_channel(0);
3252        let (resume_tx, resume_rx) = mpsc::sync_channel(0);
3253        static_files.before_cache_insert.set(move |segment| {
3254            assert_eq!(segment, StaticFileSegment::Headers);
3255            loaded_tx.send(()).expect("test controller dropped");
3256            resume_rx
3257                .recv_timeout(Duration::from_secs(10))
3258                .expect("test controller did not resume");
3259        });
3260
3261        // Pause a cache fill after it loads the header jar containing only block 0.
3262        let reader_static_files = static_files.clone();
3263        let reader = thread::spawn(move || reader_static_files.block_hash(0));
3264        loaded_rx
3265            .recv_timeout(Duration::from_secs(10))
3266            .expect("reader did not reach cache-fill hook");
3267
3268        // Publish a newer snapshot of the same header jar, then simulate unrelated segment
3269        // pruning invalidating the whole cache while the old load remains in flight.
3270        let hash_1 = B256::repeat_byte(0x11);
3271        let header_1 = Header { number: 1, ..Default::default() };
3272        {
3273            let mut writer = static_files.latest_writer(StaticFileSegment::Headers)?;
3274            writer.append_header(&header_1, &hash_1)?;
3275            writer.commit()?;
3276        }
3277        static_files.delete_jar(StaticFileSegment::Receipts, 2)?;
3278
3279        resume_tx.send(()).expect("reader dropped before cache publication");
3280        assert_eq!(reader.join().expect("reader panicked")?, Some(hash_0));
3281        assert_eq!(
3282            static_files.block_hash(1)?,
3283            Some(hash_1),
3284            "cache fill started before reinitialization repopulated an invalidated jar"
3285        );
3286
3287        Ok(())
3288    }
3289
3290    #[test]
3291    fn cache_fill_validation_is_serialized_with_index_reinitialization() -> eyre::Result<()> {
3292        let (static_dir, _) = create_test_static_files_dir();
3293        let static_files: StaticFileProvider<EthPrimitives> =
3294            StaticFileProviderBuilder::read_write(&static_dir).with_blocks_per_file(10).build()?;
3295        let hash = B256::repeat_byte(0x10);
3296        {
3297            let mut writer = static_files.latest_writer(StaticFileSegment::Headers)?;
3298            writer.append_header(&Header::default(), &hash)?;
3299            writer.commit()?;
3300        }
3301        static_files.remove_cached_provider(StaticFileSegment::Headers, 9);
3302
3303        let generation = static_files.cache_generation.load(Ordering::Acquire);
3304        let (validated_tx, validated_rx) = mpsc::sync_channel(0);
3305        let (resume_tx, resume_rx) = mpsc::sync_channel(0);
3306        static_files.after_cache_validation.set(move |segment| {
3307            assert_eq!(segment, StaticFileSegment::Headers);
3308            validated_tx.send(()).expect("test controller dropped");
3309            resume_rx
3310                .recv_timeout(Duration::from_secs(10))
3311                .expect("test controller did not resume");
3312        });
3313
3314        let reader_static_files = static_files.clone();
3315        let reader = thread::spawn(move || reader_static_files.block_hash(0));
3316        validated_rx
3317            .recv_timeout(Duration::from_secs(10))
3318            .expect("reader did not reach cache validation");
3319
3320        // The shard must remain locked between checking the generation and inserting the jar.
3321        // Otherwise a clear can finish here and the reader can repopulate the cache afterward.
3322        assert!(static_files.map.try_get(&(9, StaticFileSegment::Headers)).is_locked());
3323        let invalidator_static_files = static_files.clone();
3324        let invalidator = thread::spawn(move || invalidator_static_files.initialize_index());
3325        let deadline = Instant::now() + Duration::from_secs(10);
3326        while static_files.cache_generation.load(Ordering::Acquire) == generation {
3327            assert!(Instant::now() < deadline, "reinitialization did not invalidate the cache");
3328            thread::sleep(Duration::from_millis(1));
3329        }
3330
3331        // Invalidation has started, but clearing this shard must wait for publication and the
3332        // reader's guard to be released. The published snapshot must not survive that clear.
3333        resume_tx.send(()).expect("reader dropped before cache publication");
3334        assert_eq!(reader.join().expect("reader panicked")?, Some(hash));
3335        invalidator.join().expect("invalidator panicked")?;
3336        assert!(static_files.map.is_empty());
3337        assert_eq!(static_files.block_hash(0)?, Some(hash));
3338
3339        Ok(())
3340    }
3341
3342    #[test]
3343    fn test_find_fixed_range_with_block_index() -> eyre::Result<()> {
3344        let (static_dir, _) = create_test_static_files_dir();
3345        let sf_rw: StaticFileProvider<EthPrimitives> =
3346            StaticFileProviderBuilder::read_write(&static_dir).with_blocks_per_file(100).build()?;
3347
3348        let segment = StaticFileSegment::Headers;
3349
3350        // Test with None - should use default behavior
3351        assert_eq!(
3352            sf_rw.find_fixed_range_with_block_index(segment, None, 0),
3353            SegmentRangeInclusive::new(0, 99)
3354        );
3355        assert_eq!(
3356            sf_rw.find_fixed_range_with_block_index(segment, None, 250),
3357            SegmentRangeInclusive::new(200, 299)
3358        );
3359
3360        // Test with empty index - should fall back to default behavior
3361        assert_eq!(
3362            sf_rw.find_fixed_range_with_block_index(segment, Some(&BTreeMap::new()), 150),
3363            SegmentRangeInclusive::new(100, 199)
3364        );
3365
3366        // Create block index with existing ranges
3367        let block_index = BTreeMap::from_iter([
3368            (99, SegmentRangeInclusive::new(0, 99)),
3369            (199, SegmentRangeInclusive::new(100, 199)),
3370            (299, SegmentRangeInclusive::new(200, 299)),
3371        ]);
3372
3373        // Test blocks within existing ranges - should return the matching range
3374        assert_eq!(
3375            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 0),
3376            SegmentRangeInclusive::new(0, 99)
3377        );
3378        assert_eq!(
3379            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 50),
3380            SegmentRangeInclusive::new(0, 99)
3381        );
3382        assert_eq!(
3383            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 99),
3384            SegmentRangeInclusive::new(0, 99)
3385        );
3386        assert_eq!(
3387            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 100),
3388            SegmentRangeInclusive::new(100, 199)
3389        );
3390        assert_eq!(
3391            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 150),
3392            SegmentRangeInclusive::new(100, 199)
3393        );
3394        assert_eq!(
3395            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 199),
3396            SegmentRangeInclusive::new(100, 199)
3397        );
3398
3399        // Test blocks beyond existing ranges - should derive new ranges from the last range
3400        // Block 300 is exactly one segment after the last range
3401        assert_eq!(
3402            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 300),
3403            SegmentRangeInclusive::new(300, 399)
3404        );
3405        assert_eq!(
3406            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 350),
3407            SegmentRangeInclusive::new(300, 399)
3408        );
3409
3410        // Block 500 skips one segment (300-399)
3411        assert_eq!(
3412            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 500),
3413            SegmentRangeInclusive::new(500, 599)
3414        );
3415
3416        // Block 1000 skips many segments
3417        assert_eq!(
3418            sf_rw.find_fixed_range_with_block_index(segment, Some(&block_index), 1000),
3419            SegmentRangeInclusive::new(1000, 1099)
3420        );
3421
3422        // Test with block index having different sizes than blocks_per_file setting
3423        // This simulates the scenario where blocks_per_file was changed between runs
3424        let mixed_size_index = BTreeMap::from_iter([
3425            (49, SegmentRangeInclusive::new(0, 49)),     // 50 blocks
3426            (149, SegmentRangeInclusive::new(50, 149)),  // 100 blocks
3427            (349, SegmentRangeInclusive::new(150, 349)), // 200 blocks
3428        ]);
3429
3430        // Blocks within existing ranges should return those ranges regardless of size
3431        assert_eq!(
3432            sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 25),
3433            SegmentRangeInclusive::new(0, 49)
3434        );
3435        assert_eq!(
3436            sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 100),
3437            SegmentRangeInclusive::new(50, 149)
3438        );
3439        assert_eq!(
3440            sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 200),
3441            SegmentRangeInclusive::new(150, 349)
3442        );
3443
3444        // Block after the last range should derive using current blocks_per_file (100)
3445        // from the end of the last range (349)
3446        assert_eq!(
3447            sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 350),
3448            SegmentRangeInclusive::new(350, 449)
3449        );
3450        assert_eq!(
3451            sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 450),
3452            SegmentRangeInclusive::new(450, 549)
3453        );
3454        assert_eq!(
3455            sf_rw.find_fixed_range_with_block_index(segment, Some(&mixed_size_index), 550),
3456            SegmentRangeInclusive::new(550, 649)
3457        );
3458
3459        Ok(())
3460    }
3461}