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

This is useful when syncing to specific testnet blocks

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) =
-            BlockPoller::new_suspended(chain_id, block_source, blockhash_cache);
+        let (block_poller, start_tx) = BlockPoller::new_suspended(
+            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_clone));
+        let task = tokio::spawn(Self::task(start_rx, block_source, block_tx, debug_cutoff_height));
         (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,