diff --git a/common/nyxd-scraper/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs index 2b4e228d7c..1b616f3247 100644 --- a/common/nyxd-scraper/src/scraper/mod.rs +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -47,7 +47,10 @@ pub struct NyxdScraperBuilder { } impl NyxdScraperBuilder { - pub async fn build_and_start(self) -> Result { + pub async fn build_and_start( + self, + start_block: Option, + ) -> Result { let scraper = NyxdScraper::new(self.config).await?; let (processing_tx, processing_rx) = unbounded_channel(); @@ -90,6 +93,10 @@ impl NyxdScraperBuilder { ) .await?; + if let Some(height) = start_block { + scraper.process_block_range(Some(height), None).await?; + } + scraper.start_tasks(block_requester, block_processor, chain_subscriber); Ok(scraper) @@ -202,10 +209,10 @@ impl NyxdScraper { .await? .with_pruning(PruningOptions::nothing()); - let current_height = self.rpc_client.current_block_height().await? as u32; + let mut current_height = self.rpc_client.current_block_height().await? as u32; let last_processed = block_processor.last_process_height(); - let starting_height = match starting_height { + let mut starting_height = match starting_height { // always attempt to use whatever the user has provided Some(explicit) => explicit, None => { @@ -219,7 +226,8 @@ impl NyxdScraper { } }; - let end_height = match end_height { + let must_catch_up = end_height.is_none(); + let mut end_height = match end_height { // always attempt to use whatever the user has provided Some(explicit) => explicit, None => { @@ -234,32 +242,62 @@ impl NyxdScraper { } }; - info!( - starting_height = starting_height, - end_height = end_height, - "attempting to process block range" - ); + let mut last_processed = starting_height; - let range = (starting_height..=end_height).collect::>(); + while last_processed < current_height { + info!( + starting_height = starting_height, + end_height = end_height, + "attempting to process block range" + ); - // the most likely bottleneck here are going to be the chain queries, - // so batch multiple requests - for batch in range.chunks(4) { - let batch_result = join_all( - batch - .iter() - .map(|height| self.rpc_client.get_basic_block_details(*height)), - ) - .await; - for result in batch_result { - match result { - Ok(block) => block_processor.process_block(block.into()).await?, - Err(err) => { - error!("failed to retrieve the block: {err}. stopping..."); - return Err(err); + let range = (starting_height..=end_height).collect::>(); + + // the most likely bottleneck here are going to be the chain queries, + // so batch multiple requests + for batch in range.chunks(4) { + let batch_result = join_all( + batch + .iter() + .map(|height| self.rpc_client.get_basic_block_details(*height)), + ) + .await; + for result in batch_result { + match result { + Ok(block) => block_processor.process_block(block.into()).await?, + Err(err) => { + error!("failed to retrieve the block: {err}. stopping..."); + return Err(err); + } } } } + + // if we don't need to catch up, return early + if !must_catch_up { + return Ok(()); + } + + // check if we have caught up to the current block height + last_processed = end_height; + current_height = self.rpc_client.current_block_height().await? as u32; + + info!( + last_processed = last_processed, + current_height = current_height, + "🏃 still need to catch up..." + ); + + starting_height = last_processed + 1; + end_height = current_height; + } + + if must_catch_up { + info!( + last_processed = last_processed, + current_height = current_height, + "✅ block processing has caught up!" + ); } Ok(()) diff --git a/nym-data-observatory/src/chain_scraper/mod.rs b/nym-data-observatory/src/chain_scraper/mod.rs index 157dbd28bb..650b3f0c34 100644 --- a/nym-data-observatory/src/chain_scraper/mod.rs +++ b/nym-data-observatory/src/chain_scraper/mod.rs @@ -8,6 +8,10 @@ pub(crate) async fn run_chain_scraper() -> anyhow::Result { let websocket_url = reqwest::Url::parse(&websocket_url)?; let rpc_url = reqwest::Url::parse(&rpc_url)?; + let start_block_height = std::env::var("NYXD_SCRAPER_START_HEIGHT") + .ok() + .and_then(|value| value.parse::().ok()); + let scraper = NyxdScraper::builder(Config { websocket_url, rpc_url, @@ -16,7 +20,7 @@ pub(crate) async fn run_chain_scraper() -> anyhow::Result { store_precommits: false, }); - let storage = scraper.build_and_start().await?; + let instance = scraper.build_and_start(start_block_height).await?; - Ok(storage.storage) + Ok(instance.storage) }