Merge pull request #77 from hl-archive-node/feat/cutoff-latest

feat: Add debug CLI flag to enforce latest blocks (--debug-cutoff-height)

src/main.rs

@@ -12,6 +12,7 @@ use reth_hl::{
     chainspec::{parser::HlChainSpecParser, HlChainSpec},
     node::{
         cli::{Cli, HlNodeArgs},
+        rpc::precompile::{HlBlockPrecompileApiServer, HlBlockPrecompileExt},
         storage::tables::Tables,
         HlNode,
     },
@@ -39,7 +40,8 @@ fn main() -> eyre::Result<()> {
          ext: HlNodeArgs| async move {
             let default_upstream_rpc_url = builder.config().chain.official_rpc_url();
 
-            let (node, engine_handle_tx) = HlNode::new(ext.block_source_args.parse().await?);
+            let (node, engine_handle_tx) =
+                HlNode::new(ext.block_source_args.parse().await?, ext.debug_cutoff_height);
             let NodeHandle { node, node_exit_future: exit_future } = builder
                 .node(node)
                 .extend_rpc_modules(move |mut ctx| {
@@ -72,6 +74,10 @@ fn main() -> eyre::Result<()> {
                         info!("eth_getProof is disabled by default");
                     }
 
+                    ctx.modules.merge_configured(
+                        HlBlockPrecompileExt::new(ctx.registry.eth_api().clone()).into_rpc(),
+                    )?;
+
                     Ok(())
                 })
                 .apply(|builder| {

src/node/cli.rs

@@ -35,6 +35,12 @@ pub struct HlNodeArgs {
     #[command(flatten)]
     pub block_source_args: BlockSourceArgs,
 
+    /// Debug cutoff height.
+    ///
+    /// This option is used to cut off the block import at a specific height.
+    #[arg(long, env = "DEBUG_CUTOFF_HEIGHT")]
+    pub debug_cutoff_height: Option<u64>,
+
     /// Upstream RPC URL to forward incoming transactions.
     ///
     /// Default to Hyperliquid's RPC URL when not provided (https://rpc.hyperliquid.xyz/evm).

src/node/mod.rs

@@ -49,14 +49,23 @@ pub type HlNodeAddOns<N> =
 pub struct HlNode {
     engine_handle_rx: Arc<Mutex<Option<oneshot::Receiver<ConsensusEngineHandle<HlPayloadTypes>>>>>,
     block_source_config: BlockSourceConfig,
+    debug_cutoff_height: Option<u64>,
 }
 
 impl HlNode {
     pub fn new(
         block_source_config: BlockSourceConfig,
+        debug_cutoff_height: Option<u64>,
     ) -> (Self, oneshot::Sender<ConsensusEngineHandle<HlPayloadTypes>>) {
         let (tx, rx) = oneshot::channel();
-        (Self { engine_handle_rx: Arc::new(Mutex::new(Some(rx))), block_source_config }, tx)
+        (
+            Self {
+                engine_handle_rx: Arc::new(Mutex::new(Some(rx))),
+                block_source_config,
+                debug_cutoff_height,
+            },
+            tx,
+        )
     }
 }
 
@@ -84,6 +93,7 @@ impl HlNode {
             .network(HlNetworkBuilder {
                 engine_handle_rx: self.engine_handle_rx.clone(),
                 block_source_config: self.block_source_config.clone(),
+                debug_cutoff_height: self.debug_cutoff_height,
             })
             .consensus(HlConsensusBuilder::default())
     }

src/node/network/mod.rs

@@ -142,6 +142,8 @@ pub struct HlNetworkBuilder {
         Arc<Mutex<Option<oneshot::Receiver<ConsensusEngineHandle<HlPayloadTypes>>>>>,
 
     pub(crate) block_source_config: BlockSourceConfig,
+
+    pub(crate) debug_cutoff_height: Option<u64>,
 }
 
 impl HlNetworkBuilder {
@@ -203,6 +205,7 @@ where
         pool: Pool,
     ) -> eyre::Result<Self::Network> {
         let block_source_config = self.block_source_config.clone();
+        let debug_cutoff_height = self.debug_cutoff_height;
         let handle =
             ctx.start_network(NetworkManager::builder(self.network_config(ctx)?).await?, pool);
         let local_node_record = handle.local_node_record();
@@ -223,6 +226,7 @@ where
                 block_source_config
                     .create_cached_block_source((*chain_spec).clone(), next_block_number)
                     .await,
+                debug_cutoff_height,
             )
             .await
             .unwrap();

src/node/rpc/call.rs

@@ -61,7 +61,7 @@ where
         DB: Database<Error = ProviderError> + fmt::Debug,
     {
         let block_number = evm_env.block_env().number;
-        let hl_extras = self.get_hl_extras(block_number.try_into().unwrap())?;
+        let hl_extras = self.get_hl_extras(block_number.to::<u64>().into())?;
 
         let mut evm = self.evm_config().evm_with_env(db, evm_env);
         apply_precompiles(&mut evm, &hl_extras);
@@ -82,7 +82,7 @@ where
         I: InspectorFor<Self::Evm, DB>,
     {
         let block_number = evm_env.block_env().number;
-        let hl_extras = self.get_hl_extras(block_number.try_into().unwrap())?;
+        let hl_extras = self.get_hl_extras(block_number.to::<u64>().into())?;
 
         let mut evm = self.evm_config().evm_with_env_and_inspector(db, evm_env, inspector);
         apply_precompiles(&mut evm, &hl_extras);
@@ -103,7 +103,7 @@ where
         I: IntoIterator<Item = Recovered<&'a ProviderTx<Self::Provider>>>,
     {
         let block_number = evm_env.block_env().number;
-        let hl_extras = self.get_hl_extras(block_number.try_into().unwrap())?;
+        let hl_extras = self.get_hl_extras(block_number.to::<u64>().into())?;
 
         let mut evm = self.evm_config().evm_with_env(db, evm_env);
         apply_precompiles(&mut evm, &hl_extras);

src/node/rpc/estimate.rs

@@ -98,7 +98,7 @@ where
         tx_env.set_gas_limit(tx_env.gas_limit().min(highest_gas_limit));
 
         let block_number = evm_env.block_env().number;
-        let hl_extras = self.get_hl_extras(block_number.try_into().unwrap())?;
+        let hl_extras = self.get_hl_extras(block_number.to::<u64>().into())?;
 
         let mut evm = self.evm_config().evm_with_env(&mut db, evm_env);
         apply_precompiles(&mut evm, &hl_extras);

src/node/rpc/mod.rs

@@ -3,6 +3,7 @@ use crate::{
     node::{evm::apply_precompiles, types::HlExtras},
     HlBlock, HlPrimitives,
 };
+use alloy_eips::BlockId;
 use alloy_evm::Evm;
 use alloy_network::Ethereum;
 use alloy_primitives::U256;
@@ -26,7 +27,9 @@ use reth::{
 };
 use reth_evm::{ConfigureEvm, Database, EvmEnvFor, HaltReasonFor, InspectorFor, TxEnvFor};
 use reth_primitives::NodePrimitives;
-use reth_provider::{BlockReader, ChainSpecProvider, ProviderError, ProviderHeader, ProviderTx};
+use reth_provider::{
+    BlockReaderIdExt, ChainSpecProvider, ProviderError, ProviderHeader, ProviderTx,
+};
 use reth_rpc::RpcTypes;
 use reth_rpc_eth_api::{
     helpers::{
@@ -43,6 +46,7 @@ mod block;
 mod call;
 pub mod engine_api;
 mod estimate;
+pub mod precompile;
 mod transaction;
 
 pub trait HlRpcNodeCore: RpcNodeCore<Primitives: NodePrimitives<Block = HlBlock>> {}
@@ -232,7 +236,7 @@ where
         I: InspectorFor<Self::Evm, DB>,
     {
         let block_number = evm_env.block_env().number;
-        let hl_extras = self.get_hl_extras(block_number.try_into().unwrap())?;
+        let hl_extras = self.get_hl_extras(block_number.to::<u64>().into())?;
 
         let mut evm = self.evm_config().evm_with_env_and_inspector(db, evm_env, inspector);
         apply_precompiles(&mut evm, &hl_extras);
@@ -245,10 +249,10 @@ where
     N: HlRpcNodeCore,
     Rpc: RpcConvert<Primitives = N::Primitives, Error = EthApiError>,
 {
-    fn get_hl_extras(&self, block_number: u64) -> Result<HlExtras, ProviderError> {
+    fn get_hl_extras(&self, block: BlockId) -> Result<HlExtras, ProviderError> {
         Ok(self
             .provider()
-            .block_by_number(block_number)?
+            .block_by_id(block)?
             .map(|block| HlExtras {
                 read_precompile_calls: block.body.read_precompile_calls.clone(),
                 highest_precompile_address: block.body.highest_precompile_address,

src/node/rpc/precompile.rs

@@ -0,0 +1,44 @@
+use alloy_eips::BlockId;
+use jsonrpsee::proc_macros::rpc;
+use jsonrpsee_core::{async_trait, RpcResult};
+use reth_rpc_convert::RpcConvert;
+use reth_rpc_eth_types::EthApiError;
+use tracing::trace;
+
+use crate::node::{
+    rpc::{HlEthApi, HlRpcNodeCore},
+    types::HlExtras,
+};
+
+/// A custom RPC trait for fetching block precompile data.
+#[rpc(server, namespace = "eth")]
+#[async_trait]
+pub trait HlBlockPrecompileApi {
+    /// Fetches precompile data for a given block.
+    #[method(name = "blockPrecompileData")]
+    async fn block_precompile_data(&self, block: BlockId) -> RpcResult<HlExtras>;
+}
+
+pub struct HlBlockPrecompileExt<N: HlRpcNodeCore, Rpc: RpcConvert> {
+    eth_api: HlEthApi<N, Rpc>,
+}
+
+impl<N: HlRpcNodeCore, Rpc: RpcConvert> HlBlockPrecompileExt<N, Rpc> {
+    /// Creates a new instance of the [`HlBlockPrecompileExt`].
+    pub fn new(eth_api: HlEthApi<N, Rpc>) -> Self {
+        Self { eth_api }
+    }
+}
+
+#[async_trait]
+impl<N, Rpc> HlBlockPrecompileApiServer for HlBlockPrecompileExt<N, Rpc>
+where
+    N: HlRpcNodeCore,
+    Rpc: RpcConvert<Primitives = N::Primitives, Error = EthApiError>,
+{
+    async fn block_precompile_data(&self, block: BlockId) -> RpcResult<HlExtras> {
+        trace!(target: "rpc::eth", ?block, "Serving eth_blockPrecompileData");
+        let hl_extras = self.eth_api.get_hl_extras(block).map_err(EthApiError::from)?;
+        Ok(hl_extras)
+    }
+}

src/pseudo_peer/mod.rs

@@ -37,6 +37,7 @@ pub async fn start_pseudo_peer(
     chain_spec: Arc<HlChainSpec>,
     destination_peer: String,
     block_source: BlockSourceBoxed,
+    debug_cutoff_height: Option<u64>,
 ) -> eyre::Result<()> {
     let blockhash_cache = new_blockhash_cache();
 
@@ -46,6 +47,7 @@ pub async fn start_pseudo_peer(
         destination_peer,
         block_source.clone(),
         blockhash_cache.clone(),
+        debug_cutoff_height,
     )
     .await?;
 

src/pseudo_peer/network.rs

@@ -20,6 +20,7 @@ pub struct NetworkBuilder {
     discovery_port: u16,
     listener_port: u16,
     chain_spec: HlChainSpec,
+    debug_cutoff_height: Option<u64>,
 }
 
 impl Default for NetworkBuilder {
@@ -31,6 +32,7 @@ impl Default for NetworkBuilder {
             discovery_port: 0,
             listener_port: 0,
             chain_spec: HlChainSpec::default(),
+            debug_cutoff_height: None,
         }
     }
 }
@@ -46,6 +48,11 @@ impl NetworkBuilder {
         self
     }
 
+    pub fn with_debug_cutoff_height(mut self, debug_cutoff_height: Option<u64>) -> Self {
+        self.debug_cutoff_height = debug_cutoff_height;
+        self
+    }
+
     pub async fn build<BS>(
         self,
         block_source: Arc<Box<dyn super::sources::BlockSource>>,
@@ -58,8 +65,12 @@ impl NetworkBuilder {
             .listener_addr(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), self.listener_port));
         let chain_id = self.chain_spec.inner.chain().id();
 
-        let (block_poller, start_tx) =
+        let (block_poller, start_tx) = BlockPoller::new_suspended(
-            BlockPoller::new_suspended(chain_id, block_source, blockhash_cache);
+            chain_id,
+            block_source,
+            blockhash_cache,
+            self.debug_cutoff_height,
+        );
         let config = builder.block_import(Box::new(block_poller)).build(Arc::new(NoopProvider::<
             HlChainSpec,
             HlPrimitives,
@@ -77,10 +88,12 @@ pub async fn create_network_manager<BS>(
     destination_peer: String,
     block_source: Arc<Box<dyn super::sources::BlockSource>>,
     blockhash_cache: BlockHashCache,
+    debug_cutoff_height: Option<u64>,
 ) -> eyre::Result<(NetworkManager<HlNetworkPrimitives>, mpsc::Sender<()>)> {
     NetworkBuilder::default()
         .with_boot_nodes(vec![TrustedPeer::from_str(&destination_peer).unwrap()])
         .with_chain_spec(chain_spec)
+        .with_debug_cutoff_height(debug_cutoff_height)
         .build::<BS>(block_source, blockhash_cache)
         .await
 }

src/pseudo_peer/service.rs

@@ -52,12 +52,12 @@ impl BlockPoller {
         chain_id: u64,
         block_source: BS,
         blockhash_cache: BlockHashCache,
+        debug_cutoff_height: Option<u64>,
     ) -> (Self, mpsc::Sender<()>) {
         let block_source = Arc::new(block_source);
         let (start_tx, start_rx) = mpsc::channel(1);
         let (block_tx, block_rx) = mpsc::channel(100);
-        let block_tx_clone = block_tx.clone();
+        let task = tokio::spawn(Self::task(start_rx, block_source, block_tx, debug_cutoff_height));
-        let task = tokio::spawn(Self::task(start_rx, block_source, block_tx_clone));
         (Self { chain_id, block_rx, task, blockhash_cache: blockhash_cache.clone() }, start_tx)
     }
 
@@ -69,7 +69,8 @@ impl BlockPoller {
     async fn task<BS: BlockSource>(
         mut start_rx: mpsc::Receiver<()>,
         block_source: Arc<BS>,
-        block_tx_clone: mpsc::Sender<(u64, BlockAndReceipts)>,
+        block_tx: mpsc::Sender<(u64, BlockAndReceipts)>,
+        debug_cutoff_height: Option<u64>,
     ) -> eyre::Result<()> {
         start_rx.recv().await.ok_or(eyre::eyre!("Failed to receive start signal"))?;
         info!("Starting block poller");
@@ -80,10 +81,16 @@ impl BlockPoller {
             .await
             .ok_or(eyre::eyre!("Failed to find latest block number"))?;
 
+        if let Some(debug_cutoff_height) = debug_cutoff_height {
+            if 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_clone.send((next_block_number, block)).await?;
+                    block_tx.send((next_block_number, block)).await?;
                     next_block_number += 1;
                 }
                 Err(_) => tokio::time::sleep(polling_interval).await,