reth_provider/providers/database/
mod.rs

1use crate::{
2    providers::{
3        state::latest::LatestStateProvider, NodeTypesForProvider, RocksDBProvider,
4        StaticFileProvider, StaticFileProviderRWRefMut,
5    },
6    to_range,
7    traits::{BlockSource, ReceiptProvider},
8    BalProvider, BalStoreHandle, BlockHashReader, BlockNumReader, BlockReader, ChainSpecProvider,
9    DatabaseProviderFactory, EitherWriterDestination, HashedPostStateProvider, HeaderProvider,
10    HeaderSyncGapProvider, InMemoryBalStore, MetadataProvider, ProviderError,
11    PruneCheckpointReader, RocksDBProviderFactory, StageCheckpointReader, StateProviderBox,
12    StaticFileProviderFactory, StaticFileWriter, TransactionVariant, TransactionsProvider,
13};
14use alloy_consensus::transaction::TransactionMeta;
15use alloy_eips::BlockHashOrNumber;
16use alloy_primitives::{Address, BlockHash, BlockNumber, TxHash, TxNumber, B256};
17use core::fmt;
18use notify::{RecommendedWatcher, RecursiveMode, Watcher};
19use parking_lot::RwLock;
20use reth_chainspec::ChainInfo;
21use reth_db::{init_db, mdbx::DatabaseArguments, DatabaseEnv};
22use reth_db_api::{database::Database, models::StoredBlockBodyIndices};
23use reth_errors::{RethError, RethResult};
24use reth_node_types::{
25    BlockTy, HeaderTy, NodeTypesWithDB, NodeTypesWithDBAdapter, ReceiptTy, TxTy,
26};
27use reth_primitives_traits::{RecoveredBlock, SealedHeader};
28use reth_prune_types::{PruneCheckpoint, PruneModes, PruneSegment, MINIMUM_UNWIND_SAFE_DISTANCE};
29use reth_stages_types::{PipelineTarget, StageCheckpoint, StageId};
30use reth_static_file_types::StaticFileSegment;
31use reth_storage_api::{
32    BlockBodyIndicesProvider, ChainStateBlockReader, ChainStateBlockWriter, DBProvider,
33    NodePrimitivesProvider, StorageSettings, StorageSettingsCache, TryIntoHistoricalStateProvider,
34};
35use reth_storage_errors::provider::ProviderResult;
36use reth_trie::HashedPostState;
37use reth_trie_db::ChangesetCache;
38use revm::database::BundleState;
39use std::{
40    ops::{RangeBounds, RangeInclusive},
41    path::Path,
42    sync::{
43        atomic::{AtomicU64, Ordering},
44        Arc, Mutex,
45    },
46};
47use tracing::{info, instrument, trace, warn};
48
49mod provider;
50pub use provider::{
51    CommitOrder, DatabaseProvider, DatabaseProviderRO, DatabaseProviderRW, SaveBlocksMode,
52};
53
54use super::ProviderNodeTypes;
55use reth_trie::KeccakKeyHasher;
56
57mod builder;
58pub use builder::{ProviderFactoryBuilder, ReadOnlyConfig};
59
60mod metrics;
61pub use metrics::DatabaseProviderMetrics;
62
63mod chain;
64pub use chain::*;
65
66/// Sync state for read-only [`ProviderFactory`] instances.
67struct ReadOnlySyncState {
68    /// Last MDBX txn ID we synced `RocksDB` secondary / static file indexes to.
69    last_synced_txnid: AtomicU64,
70    /// Serializes the slow-path catch-up (`RocksDB` + static file re-init).
71    sync_lock: Mutex<()>,
72}
73
74/// A common provider that fetches data from a database or static file.
75///
76/// This provider implements most provider or provider factory traits.
77pub struct ProviderFactory<N: NodeTypesWithDB> {
78    /// Database instance
79    db: N::DB,
80    /// Chain spec
81    chain_spec: Arc<N::ChainSpec>,
82    /// Static File Provider
83    static_file_provider: StaticFileProvider<N::Primitives>,
84    /// Optional pruning configuration
85    prune_modes: PruneModes,
86    /// The node storage handler.
87    storage: Arc<N::Storage>,
88    /// Storage configuration settings for this node
89    storage_settings: Arc<RwLock<StorageSettings>>,
90    /// `RocksDB` provider
91    rocksdb_provider: RocksDBProvider,
92    /// Changeset cache for trie unwinding
93    changeset_cache: ChangesetCache,
94    /// Store for block access lists.
95    bal_store: BalStoreHandle,
96    /// Task runtime for spawning parallel I/O work.
97    runtime: reth_tasks::Runtime,
98    /// Minimum distance from tip required before pruning can occur.
99    minimum_pruning_distance: u64,
100    /// Database provider metrics shared by providers created from this factory.
101    database_provider_metrics: Arc<DatabaseProviderMetrics>,
102    /// State for on-demand syncing of `RocksDB` secondary and static file indexes.
103    ///
104    /// Only set for read-only factories. Can be disabled if there is no concurrent read-write
105    /// factory writing to the database (e.g as part of a running reth node).
106    read_only_sync: Option<Arc<ReadOnlySyncState>>,
107}
108
109impl<N: NodeTypesForProvider> ProviderFactory<NodeTypesWithDBAdapter<N, DatabaseEnv>> {
110    /// Instantiates the builder for this type
111    pub fn builder() -> ProviderFactoryBuilder<N> {
112        ProviderFactoryBuilder::default()
113    }
114}
115
116impl<N: ProviderNodeTypes> ProviderFactory<N> {
117    /// Create new database provider factory.
118    ///
119    /// The storage backends used by the produced factory MAY be inconsistent.
120    /// It is recommended to call [`Self::check_consistency`] after
121    /// creation to ensure consistency between the database and static files.
122    /// If the function returns unwind targets, the caller MUST unwind the
123    /// inner database to the minimum of the two targets to ensure consistency.
124    pub fn new(
125        db: N::DB,
126        chain_spec: Arc<N::ChainSpec>,
127        static_file_provider: StaticFileProvider<N::Primitives>,
128        rocksdb_provider: RocksDBProvider,
129        runtime: reth_tasks::Runtime,
130    ) -> ProviderResult<Self> {
131        // Load storage settings from database at init time. Creates a temporary provider
132        // to read persisted settings, falling back to legacy defaults if none exist.
133        //
134        // Both factory and all providers it creates should share these cached settings.
135        let legacy_settings = StorageSettings::v1();
136        let database_provider_metrics = Arc::new(DatabaseProviderMetrics::default());
137        let storage_settings = DatabaseProvider::<_, N>::new(
138            db.tx()?,
139            chain_spec.clone(),
140            static_file_provider.clone(),
141            Default::default(),
142            Default::default(),
143            Arc::new(RwLock::new(legacy_settings)),
144            rocksdb_provider.clone(),
145            ChangesetCache::new(),
146            runtime.clone(),
147            db.path(),
148            database_provider_metrics.clone(),
149        )
150        .storage_settings()?
151        .unwrap_or(legacy_settings);
152
153        Ok(Self {
154            db,
155            chain_spec,
156            static_file_provider,
157            prune_modes: PruneModes::default(),
158            storage: Default::default(),
159            storage_settings: Arc::new(RwLock::new(storage_settings)),
160            rocksdb_provider,
161            changeset_cache: ChangesetCache::new(),
162            bal_store: BalStoreHandle::new(InMemoryBalStore::default()),
163            runtime,
164            minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE,
165            database_provider_metrics,
166            read_only_sync: None,
167        })
168    }
169
170    /// Create new database provider factory and perform consistency checks.
171    ///
172    /// This will call [`Self::check_consistency`] internally and return
173    /// [`ProviderError::MustUnwind`] if inconsistencies are found. It may also
174    /// return any [`ProviderError`] that [`Self::new`] may return, or that are
175    /// encountered during consistency checks.
176    pub fn new_checked(
177        db: N::DB,
178        chain_spec: Arc<N::ChainSpec>,
179        static_file_provider: StaticFileProvider<N::Primitives>,
180        rocksdb_provider: RocksDBProvider,
181        runtime: reth_tasks::Runtime,
182    ) -> ProviderResult<Self> {
183        Self::new(db, chain_spec, static_file_provider, rocksdb_provider, runtime)
184            .and_then(Self::assert_consistent)
185    }
186}
187
188impl<N: NodeTypesWithDB> ProviderFactory<N> {
189    /// Sets the pruning configuration for an existing [`ProviderFactory`].
190    pub fn with_prune_modes(mut self, prune_modes: PruneModes) -> Self {
191        self.prune_modes = prune_modes;
192        self
193    }
194
195    /// Sets the BAL store for an existing [`ProviderFactory`].
196    pub fn with_bal_store(mut self, bal_store: BalStoreHandle) -> Self {
197        self.bal_store = bal_store;
198        self
199    }
200
201    /// Sets the changeset cache for an existing [`ProviderFactory`].
202    pub fn with_changeset_cache(mut self, changeset_cache: ChangesetCache) -> Self {
203        self.changeset_cache = changeset_cache;
204        self
205    }
206
207    /// Sets the minimum pruning distance for an existing [`ProviderFactory`].
208    ///
209    /// This controls the minimum distance from tip required before pruning can occur.
210    /// The default is [`MINIMUM_UNWIND_SAFE_DISTANCE`].
211    pub const fn with_minimum_pruning_distance(mut self, distance: u64) -> Self {
212        self.minimum_pruning_distance = distance;
213        self
214    }
215
216    /// Enables on-demand syncing of `RocksDB` secondary and static file indexes for read-only
217    /// factories. Initializes the tracker to the current MDBX txn ID.
218    ///
219    /// Should be used for read-only factories that are running concurrently to a reth node writing
220    /// new data to the database. Would effectively be a no-op if database directory is unchanged.
221    pub fn with_read_only_sync(mut self, watch: bool) -> Self
222    where
223        N::DB: Database,
224    {
225        // Initialize to 0 so the first `sync_providers_if_needed` call always
226        // triggers a RocksDB/static-file catch-up, regardless of what MDBX txnid
227        // the database was at when we opened it.
228        let state = Arc::new(ReadOnlySyncState {
229            last_synced_txnid: AtomicU64::new(0),
230            sync_lock: Mutex::new(()),
231        });
232        self.read_only_sync = Some(state);
233
234        if watch {
235            self.watch_db_directory();
236        }
237        self
238    }
239
240    /// Watches the MDBX data directory for changes and eagerly syncs `RocksDB` secondary and
241    /// static file indexes when modifications are detected.
242    fn watch_db_directory(&self)
243    where
244        N::DB: Database,
245    {
246        let factory = self.clone();
247        let db_path = self.db.path();
248        reth_tasks::spawn_os_thread("ro-sync", move || {
249            let (tx, rx) = std::sync::mpsc::channel();
250            let mut watcher = RecommendedWatcher::new(
251                move |res| {
252                    let _ = tx.send(res);
253                },
254                notify::Config::default(),
255            )
256            .expect("failed to create watcher");
257
258            watcher
259                .watch(&db_path, RecursiveMode::NonRecursive)
260                .expect("failed to watch MDBX path");
261
262            while let Ok(res) = rx.recv() {
263                match res {
264                    Ok(event) => {
265                        if !matches!(
266                            event.kind,
267                            notify::EventKind::Modify(_) | notify::EventKind::Create(_)
268                        ) {
269                            continue;
270                        }
271
272                        if let Err(err) = factory.sync_providers_if_needed() {
273                            warn!(target: "reth::provider", %err, "background ro-sync failed");
274                        }
275                    }
276                    Err(err) => {
277                        warn!(target: "reth::provider", ?err, "MDBX directory watcher error");
278                    }
279                }
280            }
281        });
282    }
283
284    /// For read-only factories, checks whether the MDBX committed txn ID has advanced since the
285    /// last sync and, if so, catches up the `RocksDB` secondary instance and re-initializes the
286    /// static file index.
287    ///
288    /// No-op for read-write factories.
289    pub fn sync_providers_if_needed(&self) -> ProviderResult<()> {
290        let Some(sync_state) = &self.read_only_sync else { return Ok(()) };
291        let current_txnid = self.db.last_txnid().unwrap_or(0);
292
293        // Fast path: no contention when nothing changed.
294        if current_txnid == sync_state.last_synced_txnid.load(Ordering::Relaxed) {
295            return Ok(());
296        }
297
298        // Slow path: serialize the actual catch-up I/O.
299        let _guard = sync_state.sync_lock.lock().unwrap_or_else(|e| e.into_inner());
300
301        // Double-check after acquiring the lock — another thread may have already synced.
302        if current_txnid == sync_state.last_synced_txnid.load(Ordering::Relaxed) {
303            return Ok(());
304        }
305
306        self.rocksdb_provider.try_catch_up_with_primary()?;
307        self.static_file_provider.initialize_index()?;
308        sync_state.last_synced_txnid.store(current_txnid, Ordering::Relaxed);
309        Ok(())
310    }
311
312    /// Returns reference to the underlying database.
313    pub const fn db_ref(&self) -> &N::DB {
314        &self.db
315    }
316
317    #[cfg(any(test, feature = "test-utils"))]
318    /// Consumes Self and returns DB
319    pub fn into_db(self) -> N::DB {
320        self.db
321    }
322}
323
324impl<N: NodeTypesWithDB> StorageSettingsCache for ProviderFactory<N> {
325    fn cached_storage_settings(&self) -> StorageSettings {
326        *self.storage_settings.read()
327    }
328
329    fn set_storage_settings_cache(&self, settings: StorageSettings) {
330        *self.storage_settings.write() = settings;
331    }
332}
333
334impl<N: NodeTypesWithDB> RocksDBProviderFactory for ProviderFactory<N> {
335    fn rocksdb_provider(&self) -> RocksDBProvider {
336        self.rocksdb_provider.clone()
337    }
338
339    fn set_pending_rocksdb_batch(&self, _batch: rocksdb::WriteBatchWithTransaction<true>) {
340        unimplemented!("ProviderFactory is a factory, not a provider - use DatabaseProvider::set_pending_rocksdb_batch instead")
341    }
342
343    fn commit_pending_rocksdb_batches(&self) -> ProviderResult<()> {
344        unimplemented!("ProviderFactory is a factory, not a provider - use DatabaseProvider::commit_pending_rocksdb_batches instead")
345    }
346}
347
348impl<N: ProviderNodeTypes<DB = DatabaseEnv>> ProviderFactory<N> {
349    /// Create new database provider by passing a path. [`ProviderFactory`] will own the database
350    /// instance.
351    pub fn new_with_database_path<P: AsRef<Path>>(
352        path: P,
353        chain_spec: Arc<N::ChainSpec>,
354        args: DatabaseArguments,
355        static_file_provider: StaticFileProvider<N::Primitives>,
356        rocksdb_provider: RocksDBProvider,
357        runtime: reth_tasks::Runtime,
358    ) -> RethResult<Self> {
359        Self::new(
360            init_db(path, args).map_err(RethError::msg)?,
361            chain_spec,
362            static_file_provider,
363            rocksdb_provider,
364            runtime,
365        )
366        .map_err(RethError::Provider)
367    }
368}
369
370impl<N: ProviderNodeTypes> ProviderFactory<N> {
371    /// Returns a provider with a created `DbTx` inside, which allows fetching data from the
372    /// database using different types of providers. Example: [`HeaderProvider`]
373    /// [`BlockHashReader`]. This may fail if the inner read database transaction fails to open.
374    ///
375    /// This sets the [`PruneModes`] to [`None`], because they should only be relevant for writing
376    /// data.
377    #[track_caller]
378    pub fn provider(&self) -> ProviderResult<DatabaseProviderRO<N::DB, N>> {
379        let db_tx = self.db.tx()?;
380
381        // Sync providers after opening the database transaction to make
382        // sure that no data is pruned from rocksdb or static files.
383        //
384        // Reorg logic ensures that no data is pruned from rocksdb or static files while there is an
385        // mdbx transaction open that might rely on this data.
386        self.sync_providers_if_needed()?;
387
388        Ok(DatabaseProvider::new(
389            db_tx,
390            self.chain_spec.clone(),
391            self.static_file_provider.clone(),
392            self.prune_modes.clone(),
393            self.storage.clone(),
394            self.storage_settings.clone(),
395            self.rocksdb_provider.clone(),
396            self.changeset_cache.clone(),
397            self.runtime.clone(),
398            self.db.path(),
399            self.database_provider_metrics.clone(),
400        )
401        .with_minimum_pruning_distance(self.minimum_pruning_distance))
402    }
403
404    /// Returns a provider with a created `DbTxMut` inside, which allows fetching and updating
405    /// data from the database using different types of providers. Example: [`HeaderProvider`]
406    /// [`BlockHashReader`].  This may fail if the inner read/write database transaction fails to
407    /// open.
408    #[track_caller]
409    pub fn provider_rw(&self) -> ProviderResult<DatabaseProviderRW<N::DB, N>> {
410        Ok(DatabaseProviderRW(
411            DatabaseProvider::new_rw(
412                self.db.tx_mut()?,
413                self.chain_spec.clone(),
414                self.static_file_provider.clone(),
415                self.prune_modes.clone(),
416                self.storage.clone(),
417                self.storage_settings.clone(),
418                self.rocksdb_provider.clone(),
419                self.changeset_cache.clone(),
420                self.runtime.clone(),
421                self.db.path(),
422                self.database_provider_metrics.clone(),
423            )
424            .with_reader_txn_tracker(self.db.clone())
425            .with_minimum_pruning_distance(self.minimum_pruning_distance),
426        ))
427    }
428
429    /// Returns a provider with a created `DbTxMut` inside, configured for unwind operations.
430    /// Uses unwind commit order (MDBX first, then `RocksDB`, then static files) to allow
431    /// recovery by truncating static files on restart if interrupted.
432    ///
433    /// Unwind commits may wait for pre-existing readers to drain before finishing later
434    /// cross-store steps. Drop any long-lived read providers before committing this provider.
435    #[track_caller]
436    pub fn unwind_provider_rw(
437        &self,
438    ) -> ProviderResult<DatabaseProvider<<N::DB as Database>::TXMut, N>> {
439        Ok(DatabaseProvider::new_unwind_rw(
440            self.db.tx_mut()?,
441            self.chain_spec.clone(),
442            self.static_file_provider.clone(),
443            self.prune_modes.clone(),
444            self.storage.clone(),
445            self.storage_settings.clone(),
446            self.rocksdb_provider.clone(),
447            self.changeset_cache.clone(),
448            self.runtime.clone(),
449            self.db.path(),
450            self.database_provider_metrics.clone(),
451        )
452        .with_reader_txn_tracker(self.db.clone())
453        .with_minimum_pruning_distance(self.minimum_pruning_distance))
454    }
455
456    /// State provider for latest block
457    #[track_caller]
458    pub fn latest(&self) -> ProviderResult<StateProviderBox> {
459        trace!(target: "providers::db", "Returning latest state provider");
460        Ok(Box::new(LatestStateProvider::new(self.database_provider_ro()?)))
461    }
462
463    /// Storage provider for state at that given block
464    pub fn history_by_block_number(
465        &self,
466        block_number: BlockNumber,
467    ) -> ProviderResult<StateProviderBox> {
468        let state_provider = self.provider()?.try_into_history_at_block(block_number)?;
469        trace!(target: "providers::db", ?block_number, "Returning historical state provider for block number");
470        Ok(state_provider)
471    }
472
473    /// Storage provider for state at that given block hash
474    pub fn history_by_block_hash(&self, block_hash: BlockHash) -> ProviderResult<StateProviderBox> {
475        let provider = self.provider()?;
476
477        let block_number = provider
478            .block_number(block_hash)?
479            .ok_or(ProviderError::BlockHashNotFound(block_hash))?;
480
481        let state_provider = provider.try_into_history_at_block(block_number)?;
482        trace!(target: "providers::db", ?block_number, %block_hash, "Returning historical state provider for block hash");
483        Ok(state_provider)
484    }
485
486    /// Asserts that the static files and database are consistent. If not,
487    /// returns [`ProviderError::MustUnwind`] with the appropriate unwind
488    /// target. May also return any [`ProviderError`] that
489    /// [`Self::check_consistency`] may return.
490    pub fn assert_consistent(self) -> ProviderResult<Self> {
491        let (rocksdb_unwind, static_file_unwind) = self.check_consistency()?;
492
493        let source = match (rocksdb_unwind, static_file_unwind) {
494            (None, None) => return Ok(self),
495            (Some(_), Some(_)) => "RocksDB and Static Files",
496            (Some(_), None) => "RocksDB",
497            (None, Some(_)) => "Static Files",
498        };
499
500        Err(ProviderError::MustUnwind {
501            data_source: source,
502            unwind_to: rocksdb_unwind
503                .into_iter()
504                .chain(static_file_unwind)
505                .min()
506                .expect("at least one unwind target must be Some"),
507        })
508    }
509
510    /// Checks the consistency between the static files and the database. This
511    /// may result in static files being pruned or otherwise healed to ensure
512    /// consistency. I.e. this MAY result in writes to the static files.
513    #[instrument(err, skip(self))]
514    pub fn check_consistency(&self) -> ProviderResult<(Option<u64>, Option<u64>)> {
515        let provider_ro = self
516            .database_provider_ro()?
517            // Healing can run long-lived read transactions (e.g., iterating changesets
518            // over millions of blocks). Disable the default timeout so MDBX doesn't
519            // kill the transaction mid-heal, which causes a crash loop on startup.
520            .disable_long_read_transaction_safety();
521
522        // Step 1: heal file-level inconsistencies (no pruning)
523        self.static_file_provider().check_file_consistency(&provider_ro)?;
524
525        // Step 2: RocksDB consistency check (needs static files tx data)
526        let rocksdb_unwind = self.rocksdb_provider().check_consistency(&provider_ro)?;
527
528        // Step 3: Static file checkpoint consistency (may prune)
529        let static_file_unwind = self.static_file_provider().check_consistency(&provider_ro)?.map(
530            |target| match target {
531                PipelineTarget::Unwind(block) => block,
532                PipelineTarget::Sync(_) => unreachable!("check_consistency returns Unwind"),
533            },
534        );
535
536        // Step 4: Heal finalized/safe block numbers that may be ahead of the
537        // highest header on nodes coming from <=1.10.2.
538        //
539        // Unwinds already set it to the target block.
540        if rocksdb_unwind.is_none() && static_file_unwind.is_none() {
541            self.heal_chain_state_block_numbers(&provider_ro)?;
542        }
543
544        Ok((rocksdb_unwind, static_file_unwind))
545    }
546
547    /// If the stored finalized or safe block number is ahead of the highest
548    /// header, resets it to the highest header.
549    fn heal_chain_state_block_numbers(
550        &self,
551        provider_ro: &DatabaseProvider<<N::DB as Database>::TX, N>,
552    ) -> ProviderResult<()> {
553        let highest_header = self.last_block_number()?;
554
555        let finalized = provider_ro.last_finalized_block_number()?;
556        let safe = provider_ro.last_safe_block_number()?;
557
558        if finalized.is_none_or(|f| f <= highest_header) && safe.is_none_or(|s| s <= highest_header)
559        {
560            return Ok(());
561        }
562
563        let provider_rw = self.provider_rw()?;
564
565        if let Some(finalized) = finalized.filter(|&f| f > highest_header) {
566            info!(
567                target: "providers::db",
568                finalized,
569                highest_header,
570                "Healing finalized block number",
571            );
572            provider_rw.save_finalized_block_number(highest_header)?;
573        }
574
575        if let Some(safe) = safe.filter(|&s| s > highest_header) {
576            info!(
577                target: "providers::db",
578                safe,
579                highest_header,
580                "Healing safe block number",
581            );
582            provider_rw.save_safe_block_number(highest_header)?;
583        }
584
585        provider_rw.commit()?;
586
587        Ok(())
588    }
589
590    /// Returns a static file provider. For read-only instances, this will also invoke
591    /// [`Self::sync_providers_if_needed`] to make sure that the static file provider is up to date.
592    pub fn caught_up_static_file_provider(
593        &self,
594    ) -> ProviderResult<StaticFileProvider<N::Primitives>> {
595        self.sync_providers_if_needed()?;
596        Ok(self.static_file_provider.clone())
597    }
598}
599
600impl<N: NodeTypesWithDB> NodePrimitivesProvider for ProviderFactory<N> {
601    type Primitives = N::Primitives;
602}
603
604impl<N: NodeTypesWithDB> BalProvider for ProviderFactory<N> {
605    fn bal_store(&self) -> &BalStoreHandle {
606        &self.bal_store
607    }
608}
609
610impl<N: ProviderNodeTypes> DatabaseProviderFactory for ProviderFactory<N> {
611    type DB = N::DB;
612    type Provider = DatabaseProvider<<N::DB as Database>::TX, N>;
613    type ProviderRW = DatabaseProvider<<N::DB as Database>::TXMut, N>;
614
615    fn database_provider_ro(&self) -> ProviderResult<Self::Provider> {
616        self.provider()
617    }
618
619    fn database_provider_rw(&self) -> ProviderResult<Self::ProviderRW> {
620        self.provider_rw().map(|provider| provider.0)
621    }
622}
623
624impl<N: NodeTypesWithDB> StaticFileProviderFactory for ProviderFactory<N> {
625    /// Returns static file provider
626    fn static_file_provider(&self) -> StaticFileProvider<Self::Primitives> {
627        self.static_file_provider.clone()
628    }
629
630    fn get_static_file_writer(
631        &self,
632        block: BlockNumber,
633        segment: StaticFileSegment,
634    ) -> ProviderResult<StaticFileProviderRWRefMut<'_, Self::Primitives>> {
635        self.static_file_provider.get_writer(block, segment)
636    }
637}
638
639impl<N: ProviderNodeTypes> HeaderSyncGapProvider for ProviderFactory<N> {
640    type Header = HeaderTy<N>;
641    fn local_tip_header(
642        &self,
643        highest_uninterrupted_block: BlockNumber,
644    ) -> ProviderResult<SealedHeader<Self::Header>> {
645        self.provider()?.local_tip_header(highest_uninterrupted_block)
646    }
647}
648
649impl<N: ProviderNodeTypes> HeaderProvider for ProviderFactory<N> {
650    type Header = HeaderTy<N>;
651
652    fn header(&self, block_hash: BlockHash) -> ProviderResult<Option<Self::Header>> {
653        self.provider()?.header(block_hash)
654    }
655
656    fn header_by_number(&self, num: BlockNumber) -> ProviderResult<Option<Self::Header>> {
657        self.caught_up_static_file_provider()?.header_by_number(num)
658    }
659
660    fn headers_range(
661        &self,
662        range: impl RangeBounds<BlockNumber>,
663    ) -> ProviderResult<Vec<Self::Header>> {
664        self.caught_up_static_file_provider()?.headers_range(range)
665    }
666
667    fn sealed_header(
668        &self,
669        number: BlockNumber,
670    ) -> ProviderResult<Option<SealedHeader<Self::Header>>> {
671        self.caught_up_static_file_provider()?.sealed_header(number)
672    }
673
674    fn sealed_headers_range(
675        &self,
676        range: impl RangeBounds<BlockNumber>,
677    ) -> ProviderResult<Vec<SealedHeader<Self::Header>>> {
678        self.caught_up_static_file_provider()?.sealed_headers_range(range)
679    }
680
681    fn sealed_headers_while(
682        &self,
683        range: impl RangeBounds<BlockNumber>,
684        predicate: impl FnMut(&SealedHeader<Self::Header>) -> bool,
685    ) -> ProviderResult<Vec<SealedHeader<Self::Header>>> {
686        self.caught_up_static_file_provider()?.sealed_headers_while(range, predicate)
687    }
688}
689
690impl<N: ProviderNodeTypes> BlockHashReader for ProviderFactory<N> {
691    fn block_hash(&self, number: u64) -> ProviderResult<Option<B256>> {
692        self.caught_up_static_file_provider()?.block_hash(number)
693    }
694
695    fn canonical_hashes_range(
696        &self,
697        start: BlockNumber,
698        end: BlockNumber,
699    ) -> ProviderResult<Vec<B256>> {
700        self.caught_up_static_file_provider()?.canonical_hashes_range(start, end)
701    }
702}
703
704impl<N: ProviderNodeTypes> BlockNumReader for ProviderFactory<N> {
705    fn chain_info(&self) -> ProviderResult<ChainInfo> {
706        self.provider()?.chain_info()
707    }
708
709    fn best_block_number(&self) -> ProviderResult<BlockNumber> {
710        self.provider()?.best_block_number()
711    }
712
713    fn last_block_number(&self) -> ProviderResult<BlockNumber> {
714        self.caught_up_static_file_provider()?.last_block_number()
715    }
716
717    fn earliest_block_number(&self) -> ProviderResult<BlockNumber> {
718        // earliest history height tracks the lowest block number that has __not__ been expired, in
719        // other words, the first/earliest available block.
720        Ok(self.caught_up_static_file_provider()?.earliest_history_height())
721    }
722
723    fn block_number(&self, hash: B256) -> ProviderResult<Option<BlockNumber>> {
724        self.provider()?.block_number(hash)
725    }
726}
727
728impl<N: ProviderNodeTypes> BlockReader for ProviderFactory<N> {
729    type Block = BlockTy<N>;
730
731    fn find_block_by_hash(
732        &self,
733        hash: B256,
734        source: BlockSource,
735    ) -> ProviderResult<Option<Self::Block>> {
736        self.provider()?.find_block_by_hash(hash, source)
737    }
738
739    fn block(&self, id: BlockHashOrNumber) -> ProviderResult<Option<Self::Block>> {
740        self.provider()?.block(id)
741    }
742
743    fn pending_block(&self) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
744        self.provider()?.pending_block()
745    }
746
747    fn pending_block_and_receipts(
748        &self,
749    ) -> ProviderResult<Option<(RecoveredBlock<Self::Block>, Vec<Self::Receipt>)>> {
750        self.provider()?.pending_block_and_receipts()
751    }
752
753    fn recovered_block(
754        &self,
755        id: BlockHashOrNumber,
756        transaction_kind: TransactionVariant,
757    ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
758        self.provider()?.recovered_block(id, transaction_kind)
759    }
760
761    fn sealed_block_with_senders(
762        &self,
763        id: BlockHashOrNumber,
764        transaction_kind: TransactionVariant,
765    ) -> ProviderResult<Option<RecoveredBlock<Self::Block>>> {
766        self.provider()?.sealed_block_with_senders(id, transaction_kind)
767    }
768
769    fn block_range(&self, range: RangeInclusive<BlockNumber>) -> ProviderResult<Vec<Self::Block>> {
770        self.provider()?.block_range(range)
771    }
772
773    fn block_with_senders_range(
774        &self,
775        range: RangeInclusive<BlockNumber>,
776    ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
777        self.provider()?.block_with_senders_range(range)
778    }
779
780    fn recovered_block_range(
781        &self,
782        range: RangeInclusive<BlockNumber>,
783    ) -> ProviderResult<Vec<RecoveredBlock<Self::Block>>> {
784        self.provider()?.recovered_block_range(range)
785    }
786
787    fn block_by_transaction_id(&self, id: TxNumber) -> ProviderResult<Option<BlockNumber>> {
788        self.provider()?.block_by_transaction_id(id)
789    }
790}
791
792impl<N: ProviderNodeTypes> TransactionsProvider for ProviderFactory<N> {
793    type Transaction = TxTy<N>;
794
795    fn transaction_id(&self, tx_hash: TxHash) -> ProviderResult<Option<TxNumber>> {
796        self.provider()?.transaction_id(tx_hash)
797    }
798
799    fn transaction_by_id(&self, id: TxNumber) -> ProviderResult<Option<Self::Transaction>> {
800        self.caught_up_static_file_provider()?.transaction_by_id(id)
801    }
802
803    fn transaction_by_id_unhashed(
804        &self,
805        id: TxNumber,
806    ) -> ProviderResult<Option<Self::Transaction>> {
807        self.caught_up_static_file_provider()?.transaction_by_id_unhashed(id)
808    }
809
810    fn transaction_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Transaction>> {
811        self.provider()?.transaction_by_hash(hash)
812    }
813
814    fn transaction_by_hash_with_meta(
815        &self,
816        tx_hash: TxHash,
817    ) -> ProviderResult<Option<(Self::Transaction, TransactionMeta)>> {
818        self.provider()?.transaction_by_hash_with_meta(tx_hash)
819    }
820
821    fn transactions_by_block(
822        &self,
823        id: BlockHashOrNumber,
824    ) -> ProviderResult<Option<Vec<Self::Transaction>>> {
825        self.provider()?.transactions_by_block(id)
826    }
827
828    fn transactions_by_block_range(
829        &self,
830        range: impl RangeBounds<BlockNumber>,
831    ) -> ProviderResult<Vec<Vec<Self::Transaction>>> {
832        self.provider()?.transactions_by_block_range(range)
833    }
834
835    fn transactions_by_tx_range(
836        &self,
837        range: impl RangeBounds<TxNumber>,
838    ) -> ProviderResult<Vec<Self::Transaction>> {
839        self.caught_up_static_file_provider()?.transactions_by_tx_range(range)
840    }
841
842    fn senders_by_tx_range(
843        &self,
844        range: impl RangeBounds<TxNumber>,
845    ) -> ProviderResult<Vec<Address>> {
846        if EitherWriterDestination::senders(self).is_static_file() {
847            self.caught_up_static_file_provider()?.senders_by_tx_range(range)
848        } else {
849            self.provider()?.senders_by_tx_range(range)
850        }
851    }
852
853    fn transaction_sender(&self, id: TxNumber) -> ProviderResult<Option<Address>> {
854        if EitherWriterDestination::senders(self).is_static_file() {
855            self.caught_up_static_file_provider()?.transaction_sender(id)
856        } else {
857            self.provider()?.transaction_sender(id)
858        }
859    }
860}
861
862impl<N: ProviderNodeTypes> ReceiptProvider for ProviderFactory<N> {
863    type Receipt = ReceiptTy<N>;
864
865    fn receipt(&self, id: TxNumber) -> ProviderResult<Option<Self::Receipt>> {
866        self.caught_up_static_file_provider()?.get_with_static_file_or_database(
867            StaticFileSegment::Receipts,
868            id,
869            |static_file| static_file.receipt(id),
870            || self.provider()?.receipt(id),
871        )
872    }
873
874    fn receipt_by_hash(&self, hash: TxHash) -> ProviderResult<Option<Self::Receipt>> {
875        self.provider()?.receipt_by_hash(hash)
876    }
877
878    fn receipts_by_block(
879        &self,
880        block: BlockHashOrNumber,
881    ) -> ProviderResult<Option<Vec<Self::Receipt>>> {
882        self.provider()?.receipts_by_block(block)
883    }
884
885    fn receipts_by_tx_range(
886        &self,
887        range: impl RangeBounds<TxNumber>,
888    ) -> ProviderResult<Vec<Self::Receipt>> {
889        self.caught_up_static_file_provider()?.get_range_with_static_file_or_database(
890            StaticFileSegment::Receipts,
891            to_range(range),
892            |static_file, range, _| static_file.receipts_by_tx_range(range),
893            |range, _| self.provider()?.receipts_by_tx_range(range),
894            |_| true,
895        )
896    }
897
898    fn receipts_by_block_range(
899        &self,
900        block_range: RangeInclusive<BlockNumber>,
901    ) -> ProviderResult<Vec<Vec<Self::Receipt>>> {
902        self.provider()?.receipts_by_block_range(block_range)
903    }
904}
905
906impl<N: ProviderNodeTypes> BlockBodyIndicesProvider for ProviderFactory<N> {
907    fn block_body_indices(
908        &self,
909        number: BlockNumber,
910    ) -> ProviderResult<Option<StoredBlockBodyIndices>> {
911        self.provider()?.block_body_indices(number)
912    }
913
914    fn block_body_indices_range(
915        &self,
916        range: RangeInclusive<BlockNumber>,
917    ) -> ProviderResult<Vec<StoredBlockBodyIndices>> {
918        self.provider()?.block_body_indices_range(range)
919    }
920}
921
922impl<N: ProviderNodeTypes> StageCheckpointReader for ProviderFactory<N> {
923    fn get_stage_checkpoint(&self, id: StageId) -> ProviderResult<Option<StageCheckpoint>> {
924        self.provider()?.get_stage_checkpoint(id)
925    }
926
927    fn get_stage_checkpoint_progress(&self, id: StageId) -> ProviderResult<Option<Vec<u8>>> {
928        self.provider()?.get_stage_checkpoint_progress(id)
929    }
930    fn get_all_checkpoints(&self) -> ProviderResult<Vec<(String, StageCheckpoint)>> {
931        self.provider()?.get_all_checkpoints()
932    }
933}
934
935impl<N: NodeTypesWithDB> ChainSpecProvider for ProviderFactory<N> {
936    type ChainSpec = N::ChainSpec;
937
938    fn chain_spec(&self) -> Arc<N::ChainSpec> {
939        self.chain_spec.clone()
940    }
941}
942
943impl<N: ProviderNodeTypes> PruneCheckpointReader for ProviderFactory<N> {
944    fn get_prune_checkpoint(
945        &self,
946        segment: PruneSegment,
947    ) -> ProviderResult<Option<PruneCheckpoint>> {
948        self.provider()?.get_prune_checkpoint(segment)
949    }
950
951    fn get_prune_checkpoints(&self) -> ProviderResult<Vec<(PruneSegment, PruneCheckpoint)>> {
952        self.provider()?.get_prune_checkpoints()
953    }
954}
955
956impl<N: ProviderNodeTypes> HashedPostStateProvider for ProviderFactory<N> {
957    fn hashed_post_state(&self, bundle_state: &BundleState) -> HashedPostState {
958        HashedPostState::from_bundle_state::<KeccakKeyHasher>(bundle_state.state())
959    }
960}
961
962impl<N: ProviderNodeTypes> MetadataProvider for ProviderFactory<N> {
963    fn get_metadata(&self, key: &str) -> ProviderResult<Option<Vec<u8>>> {
964        self.provider()?.get_metadata(key)
965    }
966}
967
968impl<N> fmt::Debug for ProviderFactory<N>
969where
970    N: NodeTypesWithDB<DB: fmt::Debug, ChainSpec: fmt::Debug, Storage: fmt::Debug>,
971{
972    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
973        let Self {
974            db,
975            chain_spec,
976            static_file_provider,
977            prune_modes,
978            storage,
979            storage_settings,
980            rocksdb_provider,
981            changeset_cache,
982            bal_store,
983            runtime,
984            minimum_pruning_distance,
985            database_provider_metrics: _,
986            read_only_sync,
987        } = self;
988        f.debug_struct("ProviderFactory")
989            .field("db", &db)
990            .field("chain_spec", &chain_spec)
991            .field("static_file_provider", &static_file_provider)
992            .field("prune_modes", &prune_modes)
993            .field("storage", &storage)
994            .field("storage_settings", &*storage_settings.read())
995            .field("rocksdb_provider", &rocksdb_provider)
996            .field("changeset_cache", &changeset_cache)
997            .field("bal_store", &bal_store)
998            .field("runtime", &runtime)
999            .field("minimum_pruning_distance", &minimum_pruning_distance)
1000            .field(
1001                "read_only_sync",
1002                &read_only_sync.as_ref().map(|s| s.last_synced_txnid.load(Ordering::Relaxed)),
1003            )
1004            .finish()
1005    }
1006}
1007
1008impl<N: NodeTypesWithDB> Clone for ProviderFactory<N> {
1009    fn clone(&self) -> Self {
1010        Self {
1011            db: self.db.clone(),
1012            chain_spec: self.chain_spec.clone(),
1013            static_file_provider: self.static_file_provider.clone(),
1014            prune_modes: self.prune_modes.clone(),
1015            storage: self.storage.clone(),
1016            storage_settings: self.storage_settings.clone(),
1017            rocksdb_provider: self.rocksdb_provider.clone(),
1018            changeset_cache: self.changeset_cache.clone(),
1019            bal_store: self.bal_store.clone(),
1020            runtime: self.runtime.clone(),
1021            minimum_pruning_distance: self.minimum_pruning_distance,
1022            database_provider_metrics: self.database_provider_metrics.clone(),
1023            read_only_sync: self.read_only_sync.clone(),
1024        }
1025    }
1026}
1027
1028#[cfg(test)]
1029mod tests {
1030    use super::*;
1031    use crate::{
1032        providers::{StaticFileProvider, StaticFileWriter},
1033        test_utils::{blocks::TEST_BLOCK, create_test_provider_factory, MockNodeTypesWithDB},
1034        BlockHashReader, BlockNumReader, BlockWriter, DBProvider, HeaderSyncGapProvider,
1035        TransactionsProvider,
1036    };
1037    use alloy_primitives::{TxNumber, B256};
1038    use assert_matches::assert_matches;
1039    use reth_chainspec::ChainSpecBuilder;
1040    use reth_db::{
1041        mdbx::DatabaseArguments,
1042        test_utils::{create_test_rocksdb_dir, create_test_static_files_dir, ERROR_TEMPDIR},
1043    };
1044    use reth_db_api::tables;
1045    use reth_primitives_traits::SignerRecoverable;
1046    use reth_prune_types::{PruneMode, PruneModes};
1047    use reth_storage_errors::provider::ProviderError;
1048    use reth_testing_utils::generators::{self, random_block, random_header, BlockParams};
1049    use std::{ops::RangeInclusive, sync::Arc};
1050
1051    #[test]
1052    fn common_history_provider() {
1053        let factory = create_test_provider_factory();
1054        let _ = factory.latest();
1055    }
1056
1057    #[test]
1058    fn default_chain_info() {
1059        let factory = create_test_provider_factory();
1060        let provider = factory.provider().unwrap();
1061
1062        let chain_info = provider.chain_info().expect("should be ok");
1063        assert_eq!(chain_info.best_number, 0);
1064        assert_eq!(chain_info.best_hash, B256::ZERO);
1065    }
1066
1067    #[test]
1068    fn provider_flow() {
1069        let factory = create_test_provider_factory();
1070        let provider = factory.provider().unwrap();
1071        provider.block_hash(0).unwrap();
1072        let provider_rw = factory.provider_rw().unwrap();
1073        provider_rw.block_hash(0).unwrap();
1074        provider.block_hash(0).unwrap();
1075    }
1076
1077    #[test]
1078    fn provider_factory_with_database_path() {
1079        let chain_spec = ChainSpecBuilder::mainnet().build();
1080        let (_static_dir, static_dir_path) = create_test_static_files_dir();
1081        let (_rocksdb_dir, rocksdb_path) = create_test_rocksdb_dir();
1082        let _db_tempdir = tempfile::TempDir::new().expect(ERROR_TEMPDIR);
1083        let factory = ProviderFactory::<MockNodeTypesWithDB<DatabaseEnv>>::new_with_database_path(
1084            _db_tempdir.path(),
1085            Arc::new(chain_spec),
1086            DatabaseArguments::new(Default::default()),
1087            StaticFileProvider::read_write(static_dir_path).unwrap(),
1088            RocksDBProvider::builder(&rocksdb_path).build().unwrap(),
1089            reth_tasks::Runtime::test(),
1090        )
1091        .unwrap();
1092        let provider = factory.provider().unwrap();
1093        provider.block_hash(0).unwrap();
1094        let provider_rw = factory.provider_rw().unwrap();
1095        provider_rw.block_hash(0).unwrap();
1096        provider.block_hash(0).unwrap();
1097    }
1098
1099    #[test]
1100    fn insert_block_with_prune_modes() {
1101        let block = TEST_BLOCK.clone();
1102
1103        {
1104            let factory = create_test_provider_factory();
1105            let provider = factory.provider_rw().unwrap();
1106            assert_matches!(provider.insert_block(&block.clone().try_recover().unwrap()), Ok(_));
1107            assert_matches!(
1108                provider.transaction_sender(0), Ok(Some(sender))
1109                if sender == block.body().transactions[0].recover_signer().unwrap()
1110            );
1111            assert_matches!(
1112                provider.transaction_id(*block.body().transactions[0].tx_hash()),
1113                Ok(Some(0))
1114            );
1115        }
1116
1117        {
1118            let prune_modes = PruneModes {
1119                sender_recovery: Some(PruneMode::Full),
1120                transaction_lookup: Some(PruneMode::Full),
1121                ..PruneModes::default()
1122            };
1123            // Keep factory alive until provider is dropped to prevent TempDatabase cleanup
1124            let factory = create_test_provider_factory().with_prune_modes(prune_modes);
1125            let provider = factory.provider_rw().unwrap();
1126            assert_matches!(provider.insert_block(&block.clone().try_recover().unwrap()), Ok(_));
1127            assert_matches!(provider.transaction_sender(0), Ok(None));
1128            assert_matches!(
1129                provider.transaction_id(*block.body().transactions[0].tx_hash()),
1130                Ok(None)
1131            );
1132        }
1133    }
1134
1135    #[test]
1136    fn take_block_transaction_range_recover_senders() {
1137        let mut rng = generators::rng();
1138        let block =
1139            random_block(&mut rng, 0, BlockParams { tx_count: Some(3), ..Default::default() });
1140
1141        let tx_ranges: Vec<RangeInclusive<TxNumber>> = vec![0..=0, 1..=1, 2..=2, 0..=1, 1..=2];
1142        for range in tx_ranges {
1143            let factory = create_test_provider_factory();
1144            let provider = factory.provider_rw().unwrap();
1145
1146            assert_matches!(provider.insert_block(&block.clone().try_recover().unwrap()), Ok(_));
1147
1148            let senders = provider.take::<tables::TransactionSenders>(range.clone()).unwrap();
1149            assert_eq!(
1150                senders,
1151                range
1152                    .clone()
1153                    .map(|tx_number| (
1154                        tx_number,
1155                        block.body().transactions[tx_number as usize].recover_signer().unwrap()
1156                    ))
1157                    .collect::<Vec<_>>()
1158            );
1159
1160            let db_senders = provider.senders_by_tx_range(range);
1161            assert!(matches!(db_senders, Ok(ref v) if v.is_empty()));
1162        }
1163    }
1164
1165    #[test]
1166    fn header_sync_gap_lookup() {
1167        let factory = create_test_provider_factory();
1168        let provider = factory.provider_rw().unwrap();
1169
1170        let mut rng = generators::rng();
1171
1172        // Genesis
1173        let checkpoint = 0;
1174        let head = random_header(&mut rng, 0, None);
1175
1176        // Empty database
1177        assert_matches!(
1178            provider.local_tip_header(checkpoint),
1179            Err(ProviderError::HeaderNotFound(block_number))
1180                if block_number.as_number().unwrap() == checkpoint
1181        );
1182
1183        // Checkpoint and no gap
1184        let static_file_provider = provider.static_file_provider();
1185        let mut static_file_writer =
1186            static_file_provider.latest_writer(StaticFileSegment::Headers).unwrap();
1187        static_file_writer.append_header(head.header(), &head.hash()).unwrap();
1188        static_file_writer.commit().unwrap();
1189        drop(static_file_writer);
1190
1191        let local_head = provider.local_tip_header(checkpoint).unwrap();
1192
1193        assert_eq!(local_head, head);
1194    }
1195}
reth_provider/providers/database/mod.rs

reth_provider/providers/database/
mod.rs
