feat(snapshots): write Headers to snapshots during HeaderStage

bin/reth/src/commands/debug_cmd/execution.rs

@@ -202,10 +202,11 @@ impl Command {
         fs::create_dir_all(&db_path)?;
         let db =
             Arc::new(init_db(db_path, DatabaseArguments::default().log_level(self.db.log_level))?);
-        let provider_factory = ProviderFactory::new(db.clone(), self.chain.clone());
+        let provider_factory = ProviderFactory::new(db.clone(), self.chain.clone())
+            .with_snapshots(data_dir.snapshots_path())?;
 
         debug!(target: "reth::cli", chain=%self.chain.chain, genesis=?self.chain.genesis_hash(), "Initializing genesis");
-        init_genesis(db.clone(), self.chain.clone())?;
+        init_genesis(provider_factory.clone())?;
 
         let consensus: Arc<dyn Consensus> = Arc::new(BeaconConsensus::new(Arc::clone(&self.chain)));
 

bin/reth/src/commands/import.rs

@@ -92,11 +92,12 @@ impl ImportCommand {
         let db =
             Arc::new(init_db(db_path, DatabaseArguments::default().log_level(self.db.log_level))?);
         info!(target: "reth::cli", "Database opened");
-        let provider_factory = ProviderFactory::new(db.clone(), self.chain.clone());
+        let provider_factory = ProviderFactory::new(db.clone(), self.chain.clone())
+            .with_snapshots(data_dir.snapshots_path())?;
 
         debug!(target: "reth::cli", chain=%self.chain.chain, genesis=?self.chain.genesis_hash(), "Initializing genesis");
 
-        init_genesis(db.clone(), self.chain.clone())?;
+        init_genesis(provider_factory.clone())?;
 
         let consensus = Arc::new(BeaconConsensus::new(self.chain.clone()));
         info!(target: "reth::cli", "Consensus engine initialized");

bin/reth/src/commands/init_cmd.rs

@@ -11,6 +11,7 @@ use crate::{
 use clap::Parser;
 use reth_db::{init_db, mdbx::DatabaseArguments};
 use reth_primitives::ChainSpec;
+use reth_provider::ProviderFactory;
 use std::sync::Arc;
 use tracing::info;
 
@@ -57,7 +58,10 @@ impl InitCommand {
         info!(target: "reth::cli", "Database opened");
 
         info!(target: "reth::cli", "Writing genesis block");
-        let hash = init_genesis(db, self.chain)?;
+        let provider_factory = ProviderFactory::new(db.clone(), self.chain.clone())
+            .with_snapshots(data_dir.snapshots_path())?;
+
+        let hash = init_genesis(provider_factory)?;
 
         info!(target: "reth::cli", hash = ?hash, "Genesis block written");
         Ok(())

bin/reth/src/commands/recover/storage_tries.rs

@@ -51,7 +51,9 @@ impl Command {
         let db = Arc::new(init_db(db_path, Default::default())?);
 
         debug!(target: "reth::cli", chain=%self.chain.chain, genesis=?self.chain.genesis_hash(), "Initializing genesis");
-        init_genesis(db.clone(), self.chain.clone())?;
+        let provider_factory = ProviderFactory::new(db.clone(), self.chain.clone())
+            .with_snapshots(data_dir.snapshots_path())?;
+        init_genesis(provider_factory.clone())?;
 
         let factory = ProviderFactory::new(&db, self.chain);
         let mut provider = factory.provider_rw()?;

bin/reth/src/commands/stage/drop.rs

@@ -14,6 +14,7 @@ use reth_db::{
     database::Database, mdbx::DatabaseArguments, open_db, tables, transaction::DbTxMut, DatabaseEnv,
 };
 use reth_primitives::{fs, stage::StageId, ChainSpec};
+use reth_provider::providers::SnapshotProvider;
 use std::sync::Arc;
 use tracing::info;
 
@@ -60,6 +61,7 @@ impl Command {
             open_db(db_path.as_ref(), DatabaseArguments::default().log_level(self.db.log_level))?;
 
         let tool = DbTool::new(&db, self.chain.clone())?;
+        let snapshot_provider = Arc::new(SnapshotProvider::new(data_dir.snapshots_path())?);
 
         tool.db.update(|tx| {
             match &self.stage {
@@ -70,7 +72,7 @@ impl Command {
                     tx.clear::<tables::BlockOmmers>()?;
                     tx.clear::<tables::BlockWithdrawals>()?;
                     tx.put::<tables::SyncStage>(StageId::Bodies.to_string(), Default::default())?;
-                    insert_genesis_header::<DatabaseEnv>(tx, self.chain)?;
+                    insert_genesis_header::<DatabaseEnv>(tx, snapshot_provider, self.chain)?;
                 }
                 StageEnum::Senders => {
                     tx.clear::<tables::TxSenders>()?;
@@ -155,15 +157,15 @@ impl Command {
                         StageId::TotalDifficulty.to_string(),
                         Default::default(),
                     )?;
-                    insert_genesis_header::<DatabaseEnv>(tx, self.chain)?;
+                    insert_genesis_header::<DatabaseEnv>(tx, snapshot_provider, self.chain)?;
                 }
                 StageEnum::TxLookup => {
                     tx.clear::<tables::TxHashNumber>()?;
                     tx.put::<tables::SyncStage>(
                         StageId::TransactionLookup.to_string(),
                         Default::default(),
                     )?;
-                    insert_genesis_header::<DatabaseEnv>(tx, self.chain)?;
+                    insert_genesis_header::<DatabaseEnv>(tx, snapshot_provider, self.chain)?;
                 }
                 _ => {
                     info!("Nothing to do for stage {:?}", self.stage);

crates/node-core/src/init.rs

@@ -1,19 +1,20 @@
 //! Reth genesis initialization utility functions.
 
 use reth_db::{
-    cursor::DbCursorRO,
     database::Database,
     tables,
     transaction::{DbTx, DbTxMut},
 };
 use reth_interfaces::{db::DatabaseError, provider::ProviderResult};
 use reth_primitives::{
-    stage::StageId, Account, Bytecode, ChainSpec, Receipts, StorageEntry, B256, U256,
+    stage::StageId, Account, Bytecode, ChainSpec, Receipts, SnapshotSegment, StorageEntry, B256,
+    U256,
 };
 use reth_provider::{
     bundle_state::{BundleStateInit, RevertsInit},
-    BundleStateWithReceipts, DatabaseProviderRW, HashingWriter, HistoryWriter, OriginalValuesKnown,
-    ProviderError, ProviderFactory,
+    providers::{SnapshotProvider, SnapshotWriter},
+    BundleStateWithReceipts, ChainSpecProvider, DatabaseProviderRW, HashingWriter, HeaderProvider,
+    HistoryWriter, OriginalValuesKnown, ProviderError, ProviderFactory,
 };
 use std::{
     collections::{BTreeMap, HashMap},
@@ -46,40 +47,44 @@ impl From<DatabaseError> for InitDatabaseError {
 }
 
 /// Write the genesis block if it has not already been written
-pub fn init_genesis<DB: Database>(
-    db: Arc<DB>,
-    chain: Arc<ChainSpec>,
-) -> Result<B256, InitDatabaseError> {
+pub fn init_genesis<DB: Database>(factory: ProviderFactory<DB>) -> Result<B256, InitDatabaseError> {
+    let chain = factory.chain_spec();
     let genesis = chain.genesis();
 
     let hash = chain.genesis_hash();
 
-    let tx = db.tx()?;
-    if let Some((_, db_hash)) = tx.cursor_read::<tables::CanonicalHeaders>()?.first()? {
-        if db_hash == hash {
-            debug!("Genesis already written, skipping.");
-            return Ok(hash)
-        }
+    // Check if we already have the genesis header or if we have the wrong one.
+    match factory.sealed_header(0) {
+        Ok(None) | Err(ProviderError::MissingSnapshotBlock(SnapshotSegment::Headers, 0)) => {}
+        Ok(Some(sealed_header)) => {
+            if sealed_header.hash() == hash {
+                debug!("Genesis already written, skipping.");
+                return Ok(hash)
+            }
 
-        return Err(InitDatabaseError::GenesisHashMismatch {
-            chainspec_hash: hash,
-            database_hash: db_hash,
-        })
+            return Err(InitDatabaseError::GenesisHashMismatch {
+                chainspec_hash: hash,
+                database_hash: sealed_header.hash(),
+            })
+        }
+        Err(e) => return Err(dbg!(e).into()),
     }
 
-    drop(tx);
     debug!("Writing genesis block.");
 
     // use transaction to insert genesis header
-    let factory = ProviderFactory::new(&db, chain.clone());
     let provider_rw = factory.provider_rw()?;
     insert_genesis_hashes(&provider_rw, genesis)?;
     insert_genesis_history(&provider_rw, genesis)?;
     provider_rw.commit()?;
 
     // Insert header
-    let tx = db.tx_mut()?;
-    insert_genesis_header::<DB>(&tx, chain.clone())?;
+    let tx = factory.provider_rw()?.into_tx();
+    insert_genesis_header::<DB>(
+        &tx,
+        factory.snapshot_provider().expect("should exist"),
+        chain.clone(),
+    )?;
 
     insert_genesis_state::<DB>(&tx, genesis)?;
 
@@ -160,7 +165,7 @@ pub fn insert_genesis_state<DB: Database>(
 
 /// Inserts hashes for the genesis state.
 pub fn insert_genesis_hashes<DB: Database>(
-    provider: &DatabaseProviderRW<&DB>,
+    provider: &DatabaseProviderRW<DB>,
     genesis: &reth_primitives::Genesis,
 ) -> ProviderResult<()> {
     // insert and hash accounts to hashing table
@@ -187,7 +192,7 @@ pub fn insert_genesis_hashes<DB: Database>(
 
 /// Inserts history indices for genesis accounts and storage.
 pub fn insert_genesis_history<DB: Database>(
-    provider: &DatabaseProviderRW<&DB>,
+    provider: &DatabaseProviderRW<DB>,
     genesis: &reth_primitives::Genesis,
 ) -> ProviderResult<()> {
     let account_transitions =
@@ -208,22 +213,32 @@ pub fn insert_genesis_history<DB: Database>(
 /// Inserts header for the genesis state.
 pub fn insert_genesis_header<DB: Database>(
     tx: &<DB as Database>::TXMut,
+    snapshot_provider: Arc<SnapshotProvider>,
     chain: Arc<ChainSpec>,
 ) -> ProviderResult<()> {
     let header = chain.sealed_genesis_header();
 
-    tx.put::<tables::CanonicalHeaders>(0, header.hash)?;
+    match snapshot_provider.header_by_number(0) {
+        Ok(None) | Err(ProviderError::MissingSnapshotBlock(SnapshotSegment::Headers, 0)) => {
+            let (difficulty, hash) = (header.difficulty, header.hash);
+            let mut writer = snapshot_provider.latest_writer(SnapshotSegment::Headers)?;
+            writer.append_header(header.header, difficulty, hash)?;
+            writer.commit()?;
+        }
+        Ok(Some(_)) => {}
+        Err(e) => return Err(e),
+    }
+
     tx.put::<tables::HeaderNumbers>(header.hash, 0)?;
     tx.put::<tables::BlockBodyIndices>(0, Default::default())?;
-    tx.put::<tables::HeaderTD>(0, header.difficulty.into())?;
-    tx.put::<tables::Headers>(0, header.header)?;
 
     Ok(())
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use reth_db::cursor::DbCursorRO;
 
     use reth_db::{
         models::{storage_sharded_key::StorageShardedKey, ShardedKey},
@@ -249,38 +264,37 @@ mod tests {
 
     #[test]
     fn success_init_genesis_mainnet() {
-        let db = create_test_rw_db();
-        let genesis_hash = init_genesis(db, MAINNET.clone()).unwrap();
+        let genesis_hash =
+            init_genesis(ProviderFactory::new(create_test_rw_db(), MAINNET.clone())).unwrap();
 
         // actual, expected
         assert_eq!(genesis_hash, MAINNET_GENESIS_HASH);
     }
 
     #[test]
     fn success_init_genesis_goerli() {
-        let db = create_test_rw_db();
-        let genesis_hash = init_genesis(db, GOERLI.clone()).unwrap();
+        let genesis_hash =
+            init_genesis(ProviderFactory::new(create_test_rw_db(), GOERLI.clone())).unwrap();
 
         // actual, expected
         assert_eq!(genesis_hash, GOERLI_GENESIS_HASH);
     }
 
     #[test]
     fn success_init_genesis_sepolia() {
-        let db = create_test_rw_db();
-        let genesis_hash = init_genesis(db, SEPOLIA.clone()).unwrap();
+        let genesis_hash =
+            init_genesis(ProviderFactory::new(create_test_rw_db(), SEPOLIA.clone())).unwrap();
 
         // actual, expected
         assert_eq!(genesis_hash, SEPOLIA_GENESIS_HASH);
     }
 
     #[test]
     fn fail_init_inconsistent_db() {
-        let db = create_test_rw_db();
-        init_genesis(db.clone(), SEPOLIA.clone()).unwrap();
+        init_genesis(ProviderFactory::new(create_test_rw_db(), SEPOLIA.clone())).unwrap();
 
         // Try to init db with a different genesis block
-        let genesis_hash = init_genesis(db, MAINNET.clone());
+        let genesis_hash = init_genesis(ProviderFactory::new(create_test_rw_db(), MAINNET.clone()));
 
         assert_eq!(
             genesis_hash.unwrap_err(),
@@ -322,13 +336,15 @@ mod tests {
             ..Default::default()
         });
 
-        let db = create_test_rw_db();
-        init_genesis(db.clone(), chain_spec).unwrap();
+        let factory = ProviderFactory::new(create_test_rw_db(), chain_spec);
+        init_genesis(factory.clone()).unwrap();
+
+        let provider = factory.provider().unwrap();
 
-        let tx = db.tx().expect("failed to init tx");
+        let tx = provider.tx_ref();
 
         assert_eq!(
-            collect_table_entries::<Arc<DatabaseEnv>, tables::AccountHistory>(&tx)
+            collect_table_entries::<Arc<DatabaseEnv>, tables::AccountHistory>(tx)
                 .expect("failed to collect"),
             vec![
                 (ShardedKey::new(address_with_balance, u64::MAX), IntegerList::new([0]).unwrap()),
@@ -337,7 +353,7 @@ mod tests {
         );
 
         assert_eq!(
-            collect_table_entries::<Arc<DatabaseEnv>, tables::StorageHistory>(&tx)
+            collect_table_entries::<Arc<DatabaseEnv>, tables::StorageHistory>(tx)
                 .expect("failed to collect"),
             vec![(
                 StorageShardedKey::new(address_with_storage, storage_key, u64::MAX),

crates/node-core/src/node_config.rs

@@ -1019,7 +1019,7 @@ impl<DB: Database + DatabaseMetrics + DatabaseMetadata + 'static> NodeBuilderWit
 
         debug!(target: "reth::cli", chain=%self.config.chain.chain, genesis=?self.config.chain.genesis_hash(), "Initializing genesis");
 
-        let genesis_hash = init_genesis(Arc::clone(&self.db), self.config.chain.clone())?;
+        let genesis_hash = init_genesis(provider_factory.clone())?;
 
         info!(target: "reth::cli", "{}", DisplayHardforks::new(self.config.chain.hardforks()));