Merge pull request #91 from hl-archive-node/fix/debug-cutoff

fix: Fix --debug-cutoff-height semantics

src/addons/hl_node_compliance.rs

@@ -19,62 +19,23 @@ use alloy_rpc_types::{
     TransactionInfo,
     pubsub::{Params, SubscriptionKind},
 };
-use jsonrpsee::{PendingSubscriptionSink, SubscriptionMessage, SubscriptionSink, proc_macros::rpc};
+use jsonrpsee::{PendingSubscriptionSink, proc_macros::rpc};
 use jsonrpsee_core::{RpcResult, async_trait};
 use jsonrpsee_types::{ErrorObject, error::INTERNAL_ERROR_CODE};
 use reth::{api::FullNodeComponents, builder::rpc::RpcContext, tasks::TaskSpawner};
 use reth_primitives_traits::SignedTransaction;
 use reth_provider::{BlockIdReader, BlockReader, BlockReaderIdExt, ReceiptProvider};
-use reth_rpc::{EthFilter, EthPubSub, RpcTypes, eth::pubsub::SubscriptionSerializeError};
+use reth_rpc::{EthFilter, EthPubSub};
 use reth_rpc_eth_api::{
-    EthApiServer, EthApiTypes, EthFilterApiServer, EthPubSubApiServer, FullEthApiTypes, RpcBlock,
-    RpcConvert, RpcHeader, RpcNodeCoreExt, RpcReceipt, RpcTransaction, RpcTxReq,
-    helpers::{EthBlocks, EthTransactions, LoadReceipt},
-    transaction::ConvertReceiptInput,
+    EthApiTypes, EthFilterApiServer, EthPubSubApiServer, RpcBlock, RpcConvert, RpcReceipt,
+    RpcTransaction, helpers::EthBlocks, transaction::ConvertReceiptInput,
 };
 use reth_rpc_eth_types::EthApiError;
-use serde::Serialize;
 use std::{marker::PhantomData, sync::Arc};
-use tokio_stream::{Stream, StreamExt};
+use tokio_stream::StreamExt;
 use tracing::{Instrument, trace};
 
-use crate::{HlBlock, node::primitives::HlPrimitives};
-
-pub trait EthWrapper:
-    EthApiServer<
-        RpcTxReq<Self::NetworkTypes>,
-        RpcTransaction<Self::NetworkTypes>,
-        RpcBlock<Self::NetworkTypes>,
-        RpcReceipt<Self::NetworkTypes>,
-        RpcHeader<Self::NetworkTypes>,
-    > + FullEthApiTypes<
-        Primitives = HlPrimitives,
-        NetworkTypes: RpcTypes<TransactionResponse = alloy_rpc_types_eth::Transaction>,
-    > + RpcNodeCoreExt<Provider: BlockReader<Block = HlBlock>>
-    + EthBlocks
-    + EthTransactions
-    + LoadReceipt
-    + 'static
-{
-}
-
-impl<T> EthWrapper for T where
-    T: EthApiServer<
-            RpcTxReq<Self::NetworkTypes>,
-            RpcTransaction<Self::NetworkTypes>,
-            RpcBlock<Self::NetworkTypes>,
-            RpcReceipt<Self::NetworkTypes>,
-            RpcHeader<Self::NetworkTypes>,
-        > + FullEthApiTypes<
-            Primitives = HlPrimitives,
-            NetworkTypes: RpcTypes<TransactionResponse = alloy_rpc_types_eth::Transaction>,
-        > + RpcNodeCoreExt<Provider: BlockReader<Block = HlBlock>>
-        + EthBlocks
-        + EthTransactions
-        + LoadReceipt
-        + 'static
-{
-}
+use crate::addons::utils::{EthWrapper, new_headers_stream, pipe_from_stream};
 
 #[rpc(server, namespace = "eth")]
 #[async_trait]
@@ -387,7 +348,7 @@ where
                 )
                 .await;
             } else {
-                let _ = pubsub.handle_accepted(sink, kind, params).await;
+                let _ = pipe_from_stream(sink, new_headers_stream::<Eth>(&provider)).await;
             }
         }));
         Ok(())
@@ -412,23 +373,6 @@ fn adjust_log<Eth: EthWrapper>(mut log: Log, provider: &Eth::Provider) -> Option
     Some(log)
 }
 
-async fn pipe_from_stream<T: Serialize, St: Stream<Item = T> + Unpin>(
-    sink: SubscriptionSink,
-    mut stream: St,
-) -> Result<(), ErrorObject<'static>> {
-    loop {
-        tokio::select! {
-            _ = sink.closed() => break Ok(()),
-            maybe_item = stream.next() => {
-                let Some(item) = maybe_item else { break Ok(()) };
-                let msg = SubscriptionMessage::new(sink.method_name(), sink.subscription_id(), &item)
-                    .map_err(SubscriptionSerializeError::from)?;
-                if sink.send(msg).await.is_err() { break Ok(()); }
-            }
-        }
-    }
-}
-
 pub struct HlNodeBlockFilterHttp<Eth: EthWrapper> {
     eth_api: Arc<Eth>,
     _marker: PhantomData<Eth>,

src/addons/mod.rs

@@ -1,3 +1,5 @@
 pub mod call_forwarder;
 pub mod hl_node_compliance;
 pub mod tx_forwarder;
+pub mod subscribe_fixup;
+mod utils;

src/addons/subscribe_fixup.rs

@@ -0,0 +1,54 @@
+use crate::addons::utils::{EthWrapper, new_headers_stream, pipe_from_stream};
+use alloy_rpc_types::pubsub::{Params, SubscriptionKind};
+use async_trait::async_trait;
+use jsonrpsee::PendingSubscriptionSink;
+use jsonrpsee_types::ErrorObject;
+use reth::tasks::TaskSpawner;
+use reth_rpc::EthPubSub;
+use reth_rpc_convert::RpcTransaction;
+use reth_rpc_eth_api::{EthApiTypes, EthPubSubApiServer};
+use std::sync::Arc;
+
+pub struct SubscribeFixup<Eth: EthWrapper> {
+    pubsub: Arc<EthPubSub<Eth>>,
+    provider: Arc<Eth::Provider>,
+    subscription_task_spawner: Box<dyn TaskSpawner + 'static>,
+}
+
+#[async_trait]
+impl<Eth: EthWrapper> EthPubSubApiServer<RpcTransaction<Eth::NetworkTypes>> for SubscribeFixup<Eth>
+where
+    ErrorObject<'static>: From<<Eth as EthApiTypes>::Error>,
+{
+    async fn subscribe(
+        &self,
+        pending: PendingSubscriptionSink,
+        kind: SubscriptionKind,
+        params: Option<Params>,
+    ) -> jsonrpsee::core::SubscriptionResult {
+        let sink = pending.accept().await?;
+        let (pubsub, provider) = (self.pubsub.clone(), self.provider.clone());
+        self.subscription_task_spawner.spawn(Box::pin(async move {
+            if kind == SubscriptionKind::NewHeads {
+                let _ = pipe_from_stream(sink, new_headers_stream::<Eth>(&provider)).await;
+            } else {
+                let _ = pubsub.handle_accepted(sink, kind, params).await;
+            }
+        }));
+        Ok(())
+    }
+}
+
+impl<Eth: EthWrapper> SubscribeFixup<Eth> {
+    pub fn new(
+        pubsub: Arc<EthPubSub<Eth>>,
+        provider: Arc<Eth::Provider>,
+        subscription_task_spawner: Box<dyn TaskSpawner + 'static>,
+    ) -> Self
+    where
+        Eth: EthWrapper,
+        ErrorObject<'static>: From<Eth::Error>,
+    {
+        Self { pubsub, provider, subscription_task_spawner }
+    }
+}

src/addons/utils.rs

@@ -0,0 +1,90 @@
+use std::sync::Arc;
+
+use crate::{HlBlock, HlPrimitives};
+use alloy_primitives::U256;
+use alloy_rpc_types::Header;
+use futures::StreamExt;
+use jsonrpsee::{SubscriptionMessage, SubscriptionSink};
+use jsonrpsee_types::ErrorObject;
+use reth_primitives::SealedHeader;
+use reth_provider::{BlockReader, CanonStateSubscriptions};
+use reth_rpc::{RpcTypes, eth::pubsub::SubscriptionSerializeError};
+use reth_rpc_convert::{RpcBlock, RpcHeader, RpcReceipt, RpcTransaction, RpcTxReq};
+use reth_rpc_eth_api::{
+    EthApiServer, FullEthApiTypes, RpcNodeCoreExt,
+    helpers::{EthBlocks, EthTransactions, LoadReceipt},
+};
+use serde::Serialize;
+use tokio_stream::Stream;
+
+pub trait EthWrapper:
+    EthApiServer<
+        RpcTxReq<Self::NetworkTypes>,
+        RpcTransaction<Self::NetworkTypes>,
+        RpcBlock<Self::NetworkTypes>,
+        RpcReceipt<Self::NetworkTypes>,
+        RpcHeader<Self::NetworkTypes>,
+    > + FullEthApiTypes<
+        Primitives = HlPrimitives,
+        NetworkTypes: RpcTypes<TransactionResponse = alloy_rpc_types_eth::Transaction>,
+    > + RpcNodeCoreExt<Provider: BlockReader<Block = HlBlock>>
+    + EthBlocks
+    + EthTransactions
+    + LoadReceipt
+    + 'static
+{
+}
+
+impl<T> EthWrapper for T where
+    T: EthApiServer<
+            RpcTxReq<Self::NetworkTypes>,
+            RpcTransaction<Self::NetworkTypes>,
+            RpcBlock<Self::NetworkTypes>,
+            RpcReceipt<Self::NetworkTypes>,
+            RpcHeader<Self::NetworkTypes>,
+        > + FullEthApiTypes<
+            Primitives = HlPrimitives,
+            NetworkTypes: RpcTypes<TransactionResponse = alloy_rpc_types_eth::Transaction>,
+        > + RpcNodeCoreExt<Provider: BlockReader<Block = HlBlock>>
+        + EthBlocks
+        + EthTransactions
+        + LoadReceipt
+        + 'static
+{
+}
+
+pub(super) async fn pipe_from_stream<T: Serialize, St: Stream<Item = T> + Unpin>(
+    sink: SubscriptionSink,
+    mut stream: St,
+) -> Result<(), ErrorObject<'static>> {
+    loop {
+        tokio::select! {
+            _ = sink.closed() => break Ok(()),
+            maybe_item = stream.next() => {
+                let Some(item) = maybe_item else { break Ok(()) };
+                let msg = SubscriptionMessage::new(sink.method_name(), sink.subscription_id(), &item)
+                    .map_err(SubscriptionSerializeError::from)?;
+                if sink.send(msg).await.is_err() { break Ok(()); }
+            }
+        }
+    }
+}
+
+pub(super) fn new_headers_stream<Eth: EthWrapper>(
+    provider: &Arc<Eth::Provider>,
+) -> impl Stream<Item = Header<alloy_consensus::Header>> {
+    provider.canonical_state_stream().flat_map(|new_chain| {
+        let headers = new_chain
+            .committed()
+            .blocks_iter()
+            .map(|block| {
+                Header::from_consensus(
+                    SealedHeader::new(block.header().inner.clone(), block.hash()).into(),
+                    None,
+                    Some(U256::from(block.rlp_length())),
+                )
+            })
+            .collect::<Vec<_>>();
+        futures::stream::iter(headers)
+    })
+}

src/main.rs

@@ -1,12 +1,16 @@
 use std::sync::Arc;
 
 use clap::Parser;
-use reth::builder::{NodeBuilder, NodeHandle, WithLaunchContext};
+use reth::{
+    builder::{NodeBuilder, NodeHandle, WithLaunchContext},
+    rpc::{api::EthPubSubApiServer, eth::RpcNodeCore},
+};
 use reth_db::DatabaseEnv;
 use reth_hl::{
     addons::{
         call_forwarder::{self, CallForwarderApiServer},
         hl_node_compliance::install_hl_node_compliance,
+        subscribe_fixup::SubscribeFixup,
         tx_forwarder::{self, EthForwarderApiServer},
     },
     chainspec::{HlChainSpec, parser::HlChainSpecParser},
@@ -69,6 +73,17 @@ fn main() -> eyre::Result<()> {
                         info!("eth_getProof is disabled by default");
                     }
 
+                    // This is a temporary workaround to fix the issue with custom headers
+                    // affects `eth_subscribe[type=newHeads]`
+                    ctx.modules.replace_configured(
+                        SubscribeFixup::new(
+                            Arc::new(ctx.registry.eth_handlers().pubsub.clone()),
+                            Arc::new(ctx.registry.eth_api().provider().clone()),
+                            Box::new(ctx.node().task_executor.clone()),
+                        )
+                        .into_rpc(),
+                    )?;
+
                     ctx.modules.merge_configured(
                         HlBlockPrecompileExt::new(ctx.registry.eth_api().clone()).into_rpc(),
                     )?;

src/node/migrate.rs

@@ -1,40 +1,48 @@
 use alloy_consensus::Header;
-use alloy_primitives::{b256, hex::ToHexExt, BlockHash, B256, U256};
+use alloy_primitives::{B256, BlockHash, Bytes, U256, b256, hex::ToHexExt};
 use reth::{
-    api::{NodeTypes, NodeTypesWithDBAdapter},
+    api::NodeTypesWithDBAdapter,
     args::{DatabaseArgs, DatadirArgs},
     dirs::{ChainPath, DataDirPath},
 };
 use reth_chainspec::EthChainSpec;
 use reth_db::{
-    mdbx::{tx::Tx, RO},
+    DatabaseEnv,
+    mdbx::{RO, tx::Tx},
     models::CompactU256,
     static_file::iter_static_files,
     table::Decompress,
-    DatabaseEnv,
+    tables,
+};
+use reth_db_api::{
+    cursor::{DbCursorRO, DbCursorRW},
+    transaction::{DbTx, DbTxMut},
 };
 use reth_errors::ProviderResult;
+use reth_ethereum_primitives::EthereumReceipt;
 use reth_provider::{
-    providers::{NodeTypesForProvider, StaticFileProvider},
-    static_file::SegmentRangeInclusive,
     DatabaseProvider, ProviderFactory, ReceiptProvider, StaticFileProviderFactory,
     StaticFileSegment, StaticFileWriter,
+    providers::{NodeTypesForProvider, StaticFileProvider},
+    static_file::SegmentRangeInclusive,
 };
-use std::{marker::PhantomData, path::PathBuf, sync::Arc};
+use std::{fs::File, io::Write, path::PathBuf, sync::Arc};
 use tracing::{info, warn};
 
-use crate::{chainspec::HlChainSpec, HlHeader, HlPrimitives};
+use crate::{HlHeader, HlPrimitives, chainspec::HlChainSpec};
 
-pub(super) struct Migrator<N: NodeTypesForProvider> {
+pub(crate) trait HlNodeType:
+    NodeTypesForProvider<ChainSpec = HlChainSpec, Primitives = HlPrimitives>
+{
+}
+impl<N: NodeTypesForProvider<ChainSpec = HlChainSpec, Primitives = HlPrimitives>> HlNodeType for N {}
+
+pub(super) struct Migrator<N: HlNodeType> {
     data_dir: ChainPath<DataDirPath>,
     provider_factory: ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
-    _nt: PhantomData<N>,
 }
 
-impl<N: NodeTypesForProvider> Migrator<N>
-where
-    N: NodeTypes<ChainSpec = HlChainSpec, Primitives = HlPrimitives>,
-{
+impl<N: HlNodeType> Migrator<N> {
     const MIGRATION_PATH_SUFFIX: &'static str = "migration-tmp";
 
     pub fn new(
@@ -44,7 +52,7 @@ where
     ) -> eyre::Result<Self> {
         let data_dir = datadir.clone().resolve_datadir(chain_spec.chain());
         let provider_factory = Self::provider_factory(chain_spec, datadir, database_args)?;
-        Ok(Self { data_dir, provider_factory, _nt: PhantomData })
+        Ok(Self { data_dir, provider_factory })
     }
 
     pub fn sf_provider(&self) -> StaticFileProvider<HlPrimitives> {
@@ -66,9 +74,12 @@ where
     }
 
     fn migrate_db_inner(&self) -> eyre::Result<()> {
-        self.migrate_static_files()?;
-        self.migrate_mdbx()?;
+        let migrated_mdbx = MigratorMdbx::<N>(self).migrate_mdbx()?;
+        let migrated_static_files = MigrateStaticFiles::<N>(self).migrate_static_files()?;
+
+        if migrated_mdbx || migrated_static_files {
             info!("Database migrated successfully");
+        }
         Ok(())
     }
 
@@ -76,6 +87,116 @@ where
         self.data_dir.data_dir().join(Self::MIGRATION_PATH_SUFFIX)
     }
 
+    fn provider_factory(
+        chain_spec: HlChainSpec,
+        datadir: DatadirArgs,
+        database_args: DatabaseArgs,
+    ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>> {
+        let data_dir = datadir.clone().resolve_datadir(chain_spec.chain());
+        let db_env = reth_db::init_db(data_dir.db(), database_args.database_args())?;
+        let static_file_provider = StaticFileProvider::read_only(data_dir.static_files(), false)?;
+        let db = Arc::new(db_env);
+        Ok(ProviderFactory::new(db, Arc::new(chain_spec), static_file_provider))
+    }
+}
+
+struct MigratorMdbx<'a, N: HlNodeType>(&'a Migrator<N>);
+
+impl<'a, N: HlNodeType> MigratorMdbx<'a, N> {
+    fn migrate_mdbx(&self) -> eyre::Result<bool> {
+        // if any header is in old format, we need to migrate it, so we pick the first and last one
+        let db_env = self.0.provider_factory.provider()?;
+        let mut cursor = db_env.tx_ref().cursor_read::<tables::Headers<Bytes>>()?;
+
+        let migration_needed = {
+            let first_is_old = match cursor.first()? {
+                Some((number, header)) => using_old_header(number, &header),
+                None => false,
+            };
+            let last_is_old = match cursor.last()? {
+                Some((number, header)) => using_old_header(number, &header),
+                None => false,
+            };
+            first_is_old || last_is_old
+        };
+
+        if !migration_needed {
+            return Ok(false);
+        }
+
+        check_if_migration_enabled()?;
+
+        self.migrate_mdbx_inner()?;
+        Ok(true)
+    }
+
+    fn migrate_mdbx_inner(&self) -> eyre::Result<()> {
+        // There shouldn't be many headers in mdbx, but using file for safety
+        info!("Old database detected, migrating mdbx...");
+        let conversion_tmp = self.0.conversion_tmp_dir();
+        let tmp_path = conversion_tmp.join("headers.rmp");
+
+        if conversion_tmp.exists() {
+            std::fs::remove_dir_all(&conversion_tmp)?;
+        }
+        std::fs::create_dir_all(&conversion_tmp)?;
+
+        let count = self.export_old_headers(&tmp_path)?;
+        self.import_new_headers(tmp_path, count)?;
+        Ok(())
+    }
+
+    fn export_old_headers(&self, tmp_path: &PathBuf) -> Result<i32, eyre::Error> {
+        let db_env = self.0.provider_factory.provider()?;
+        let mut cursor_read = db_env.tx_ref().cursor_read::<tables::Headers<Bytes>>()?;
+        let mut tmp_writer = File::create(tmp_path)?;
+        let mut count = 0;
+        let old_headers = cursor_read.walk(None)?.filter_map(|row| {
+            let (block_number, header) = row.ok()?;
+            if !using_old_header(block_number, &header) {
+                None
+            } else {
+                Some((block_number, Header::decompress(&header).ok()?))
+            }
+        });
+        for (block_number, header) in old_headers {
+            let receipt =
+                db_env.receipts_by_block(block_number.into())?.expect("Receipt not found");
+            let new_header = to_hl_header(receipt, header);
+            tmp_writer.write_all(&rmp_serde::to_vec(&(block_number, new_header))?)?;
+            count += 1;
+        }
+        Ok(count)
+    }
+
+    fn import_new_headers(&self, tmp_path: PathBuf, count: i32) -> Result<(), eyre::Error> {
+        let mut tmp_reader = File::open(tmp_path)?;
+        let db_env = self.0.provider_factory.provider_rw()?;
+        let mut cursor_write = db_env.tx_ref().cursor_write::<tables::Headers<Bytes>>()?;
+        for _ in 0..count {
+            let (number, header) = rmp_serde::from_read::<_, (u64, HlHeader)>(&mut tmp_reader)?;
+            cursor_write.upsert(number, &rmp_serde::to_vec(&header)?.into())?;
+        }
+        db_env.commit()?;
+        Ok(())
+    }
+}
+
+fn check_if_migration_enabled() -> Result<(), eyre::Error> {
+    if std::env::var("EXPERIMENTAL_MIGRATE_DB").is_err() {
+        let err_msg = concat!(
+            "Detected an old database format but experimental database migration is currently disabled. ",
+            "To enable migration, set EXPERIMENTAL_MIGRATE_DB=1, or alternatively, resync your node (safest option)."
+        );
+        warn!("{}", err_msg);
+        return Err(eyre::eyre!("{}", err_msg));
+    }
+    Ok(())
+}
+
+struct MigrateStaticFiles<'a, N: HlNodeType>(&'a Migrator<N>);
+
+impl<'a, N: HlNodeType> MigrateStaticFiles<'a, N> {
     fn iterate_files_for_segment(
         &self,
         block_range: SegmentRangeInclusive,
@@ -102,8 +223,8 @@ where
 
     fn create_placeholder(&self, block_range: SegmentRangeInclusive) -> eyre::Result<()> {
         // The direction is opposite here
-        let src = self.data_dir.static_files();
-        let dst = self.conversion_tmp_dir();
+        let src = self.0.data_dir.static_files();
+        let dst = self.0.conversion_tmp_dir();
 
         for (src_path, file_name) in self.iterate_files_for_segment(block_range, &src)? {
             let dst_path = dst.join(file_name);
@@ -120,8 +241,8 @@ where
         &self,
         block_range: SegmentRangeInclusive,
     ) -> eyre::Result<()> {
-        let src = self.conversion_tmp_dir();
-        let dst = self.data_dir.static_files();
+        let src = self.0.conversion_tmp_dir();
+        let dst = self.0.data_dir.static_files();
 
         for (src_path, file_name) in self.iterate_files_for_segment(block_range, &src)? {
             let dst_path = dst.join(file_name);
@@ -133,9 +254,9 @@ where
         self.create_placeholder(block_range)
     }
 
-    fn migrate_static_files(&self) -> eyre::Result<()> {
-        let conversion_tmp = self.conversion_tmp_dir();
-        let old_path = self.data_dir.static_files();
+    fn migrate_static_files(&self) -> eyre::Result<bool> {
+        let conversion_tmp = self.0.conversion_tmp_dir();
+        let old_path = self.0.data_dir.static_files();
 
         if conversion_tmp.exists() {
             std::fs::remove_dir_all(&conversion_tmp)?;
@@ -145,13 +266,12 @@ where
         let mut all_static_files = iter_static_files(&old_path)?;
         let all_static_files =
             all_static_files.remove(&StaticFileSegment::Headers).unwrap_or_default();
-        let provider = self.provider_factory.provider()?;
 
         let mut first = true;
 
         for (block_range, _tx_ranges) in all_static_files {
-            let migration_needed = self.using_old_header(block_range.start())? ||
-                self.using_old_header(block_range.end())?;
+            let migration_needed = self.using_old_header(block_range.start())?
+                || self.using_old_header(block_range.end())?;
             if !migration_needed {
                 // Create a placeholder symlink
                 self.create_placeholder(block_range)?;
@@ -159,60 +279,34 @@ where
             }
 
             if first {
-                info!("Old database detected, migrating database...");
+                check_if_migration_enabled()?;
+
+                info!("Old database detected, migrating static files...");
                 first = false;
             }
 
-            let sf_provider = self.sf_provider();
+            let sf_provider = self.0.sf_provider();
             let sf_tmp_provider = StaticFileProvider::<HlPrimitives>::read_write(&conversion_tmp)?;
+            let provider = self.0.provider_factory.provider()?;
             let block_range_for_filename = sf_provider.find_fixed_range(block_range.start());
             migrate_single_static_file(&sf_tmp_provider, &sf_provider, &provider, block_range)?;
 
             self.move_static_files_for_segment(block_range_for_filename)?;
         }
 
-        Ok(())
-    }
-
-    fn provider_factory(
-        chain_spec: HlChainSpec,
-        datadir: DatadirArgs,
-        database_args: DatabaseArgs,
-    ) -> eyre::Result<ProviderFactory<NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>> {
-        let data_dir = datadir.clone().resolve_datadir(chain_spec.chain());
-        let db_env = reth_db::init_db(data_dir.db(), database_args.database_args())?;
-        let static_file_provider = StaticFileProvider::read_only(data_dir.static_files(), false)?;
-        let db = Arc::new(db_env);
-        Ok(ProviderFactory::new(db, Arc::new(chain_spec), static_file_provider))
-    }
-
-    fn migrate_mdbx(&self) -> eyre::Result<()> {
-        // Actually not much here, all of blocks should be in the static files
-        Ok(())
+        Ok(!first)
     }
 
     fn using_old_header(&self, number: u64) -> eyre::Result<bool> {
-        let sf_provider = self.sf_provider();
+        let sf_provider = self.0.sf_provider();
         let content = old_headers_range(&sf_provider, number..=number)?;
 
         let &[row] = &content.as_slice() else {
             warn!("No header found for block {}", number);
             return Ok(false);
         };
-        let header = &row[0];
 
-        let deserialized_old = is_old_header(header);
-        let deserialized_new = is_new_header(header);
-
-        assert!(
-            deserialized_old ^ deserialized_new,
-            "Header is not valid: {} {}\ndeserialized_old: {}\ndeserialized_new: {}",
-            number,
-            header.encode_hex(),
-            deserialized_old,
-            deserialized_new
-        );
-        Ok(deserialized_old && !deserialized_new)
+        Ok(using_old_header(number, &row[0]))
     }
 }
 
@@ -242,7 +336,7 @@ fn is_new_header(header: &[u8]) -> bool {
     rmp_serde::from_slice::<HlHeader>(header).is_ok()
 }
 
-fn migrate_single_static_file<N: NodeTypesForProvider<Primitives = HlPrimitives>>(
+fn migrate_single_static_file<N: HlNodeType>(
     sf_out: &StaticFileProvider<HlPrimitives>,
     sf_in: &StaticFileProvider<HlPrimitives>,
     provider: &DatabaseProvider<Tx<RO>, NodeTypesWithDBAdapter<N, Arc<DatabaseEnv>>>,
@@ -250,8 +344,8 @@ fn migrate_single_static_file<N: NodeTypesForProvider<Primitives = HlPrimitives>
 ) -> Result<(), eyre::Error> {
     info!("Migrating block range {}...", block_range);
 
-    // block_ranges into chunks of 100000 blocks
-    const CHUNK_SIZE: u64 = 100000;
+    // block_ranges into chunks of 50000 blocks
+    const CHUNK_SIZE: u64 = 50000;
     for chunk in (0..=block_range.end()).step_by(CHUNK_SIZE as usize) {
         let end = std::cmp::min(chunk + CHUNK_SIZE - 1, block_range.end());
         let block_range = chunk..=end;
@@ -261,11 +355,8 @@ fn migrate_single_static_file<N: NodeTypesForProvider<Primitives = HlPrimitives>
         let mut writer = sf_out.get_writer(*block_range.start(), StaticFileSegment::Headers)?;
         let new_headers = std::iter::zip(headers, receipts)
             .map(|(header, receipts)| {
-                let system_tx_count =
-                    receipts.iter().filter(|r| r.cumulative_gas_used == 0).count();
                 let eth_header = Header::decompress(&header[0]).unwrap();
-                let hl_header =
-                    HlHeader::from_ethereum_header(eth_header, &receipts, system_tx_count as u64);
+                let hl_header = to_hl_header(receipts, eth_header);
 
                 let difficulty: U256 = CompactU256::decompress(&header[1]).unwrap().into();
                 let hash = BlockHash::decompress(&header[2]).unwrap();
@@ -281,6 +372,11 @@ fn migrate_single_static_file<N: NodeTypesForProvider<Primitives = HlPrimitives>
     Ok(())
 }
 
+fn to_hl_header(receipts: Vec<EthereumReceipt>, eth_header: Header) -> HlHeader {
+    let system_tx_count = receipts.iter().filter(|r| r.cumulative_gas_used == 0).count();
+    HlHeader::from_ethereum_header(eth_header, &receipts, system_tx_count as u64)
+}
+
 fn old_headers_range(
     provider: &StaticFileProvider<HlPrimitives>,
     block_range: impl std::ops::RangeBounds<u64>,
@@ -316,3 +412,18 @@ fn to_range<R: std::ops::RangeBounds<u64>>(bounds: R) -> std::ops::Range<u64> {
 
     start..end
 }
+
+fn using_old_header(number: u64, header: &[u8]) -> bool {
+    let deserialized_old = is_old_header(header);
+    let deserialized_new = is_new_header(header);
+
+    assert!(
+        deserialized_old ^ deserialized_new,
+        "Header is not valid: {} {}\ndeserialized_old: {}\ndeserialized_new: {}",
+        number,
+        header.encode_hex(),
+        deserialized_old,
+        deserialized_new
+    );
+    deserialized_old && !deserialized_new
+}

src/pseudo_peer/service.rs

@@ -81,13 +81,13 @@ impl BlockPoller {
             .await
             .ok_or(eyre::eyre!("Failed to find latest block number"))?;
 
-        if let Some(debug_cutoff_height) = debug_cutoff_height &&
-            next_block_number > debug_cutoff_height
+        loop {
+            if let Some(debug_cutoff_height) = debug_cutoff_height
+                && next_block_number > debug_cutoff_height
             {
                 next_block_number = debug_cutoff_height;
             }
 
-        loop {
             match block_source.collect_block(next_block_number).await {
                 Ok(block) => {
                     block_tx.send((next_block_number, block)).await?;

tests/run_tests.sh

@@ -0,0 +1,49 @@
+#!/bin/bash
+
+set -e
+
+export ETH_RPC_URL="${ETH_RPC_URL:-wss://hl-archive-node.xyz}"
+
+success() {
+    echo "Success: $1"
+}
+
+fail() {
+    echo "Failed: $1"
+    exit 1
+}
+
+ensure_cmd() {
+    command -v "$1" > /dev/null 2>&1 || fail "$1 is required"
+}
+
+ensure_cmd jq
+ensure_cmd cast
+ensure_cmd wscat
+
+if [[ ! "$ETH_RPC_URL" =~ ^wss?:// ]]; then
+    fail "ETH_RPC_URL must be a websocket url"
+fi
+
+TITLE="Issue #78 - eth_getLogs should return system transactions"
+cast logs \
+    --rpc-url "$ETH_RPC_URL" \
+    --from-block 15312567 \
+    --to-block 15312570 \
+    --address 0x9fdbda0a5e284c32744d2f17ee5c74b284993463 \
+    0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef \
+    | grep -q "0x00000000000000000000000020000000000000000000000000000000000000c5" \
+    && success "$TITLE" || fail "$TITLE"
+
+TITLE="Issue #78 - eth_getBlockByNumber should return the same logsBloom as official RPC"
+OFFICIAL_RPC="https://rpc.hyperliquid.xyz/evm"
+A=$(cast block 1394092 --rpc-url "$ETH_RPC_URL" -f logsBloom | md5sum)
+B=$(cast block 1394092 --rpc-url "$OFFICIAL_RPC" -f logsBloom | md5sum)
+echo node "$A"
+echo rpc\  "$B"
+[[ "$A" == "$B" ]] && success "$TITLE" || fail "$TITLE"
+
+TITLE="eth_subscribe newHeads via wscat"
+CMD='{"jsonrpc":"2.0","id":1,"method":"eth_subscribe","params":["newHeads"]}'
+wscat -w 2 -c "$ETH_RPC_URL" -x "$CMD" | tail -1 | jq -r .params.result.nonce | grep 0x \
+    && success "$TITLE" || fail "$TITLE"