diff --git a/common/nyxd-scraper/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs index 355c6d3907..7a23bb7751 100644 --- a/common/nyxd-scraper/src/block_processor/mod.rs +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -117,6 +117,8 @@ impl BlockProcessor { let last_pruned = storage.get_pruned_height().await?; let last_pruned_height = last_pruned.try_into().unwrap_or_default(); + debug!(last_processed_height = %last_processed_height, pruned_height = %last_pruned_height, "setting up block processor..."); + Ok(BlockProcessor { config, cancel, @@ -399,12 +401,15 @@ impl BlockProcessor { // but we need it to help the compiler figure out the future is `Send` async fn startup_resync(&mut self) -> Result<(), ScraperError> { assert!(self.pending_sync.is_empty()); + info!("attempting to run startup resync..."); self.maybe_prune_storage().await?; let latest_block = self.rpc_client.current_block_height().await? as u32; + info!("obtained latest block height: {latest_block}"); if latest_block > self.last_processed_height && self.last_processed_height != 0 { + info!("we have already processed some blocks in the past - attempting to resume..."); // in case we were offline for a while, // make sure we don't request blocks we'd have to prune anyway let keep_recent = self.config.pruning_options.strategy_keep_recent(); @@ -419,7 +424,9 @@ impl BlockProcessor { // this is the first time starting up if self.last_processed_height == 0 { + info!("this is the first time starting up"); let Some(starting_height) = self.config.explicit_starting_block_height else { + info!("no starting block height set - will use the default behaviour"); // nothing to do return Ok(()); }; @@ -456,7 +463,7 @@ impl BlockProcessor { } pub(crate) async fn run(&mut self) { - info!("starting processing loop"); + info!("starting block processor processing loop"); // sure, we could be more efficient and reset it on every processed block, // but the overhead is so minimal that it doesn't matter diff --git a/common/nyxd-scraper/src/scraper/mod.rs b/common/nyxd-scraper/src/scraper/mod.rs index e0508b2c09..71cb535645 100644 --- a/common/nyxd-scraper/src/scraper/mod.rs +++ b/common/nyxd-scraper/src/scraper/mod.rs @@ -139,7 +139,7 @@ pub struct NyxdScraper { task_tracker: TaskTracker, cancel_token: CancellationToken, startup_sync: Arc, - pub storage: ScraperStorage, + storage: ScraperStorage, rpc_client: RpcClient, } @@ -163,6 +163,10 @@ impl NyxdScraper { }) } + pub fn storage(&self) -> ScraperStorage { + self.storage.clone() + } + fn start_tasks( &self, mut block_requester: BlockRequester, diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index 39a2842fa8..3b29438bbb 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -55,6 +55,12 @@ pub(crate) fn log_db_operation_time(op_name: &str, start_time: Instant) { impl ScraperStorage { #[instrument] pub async fn init + Debug>(database_path: P) -> Result { + let database_path = database_path.as_ref(); + debug!( + "initialising scraper database path to '{}'", + database_path.display() + ); + let opts = sqlx::sqlite::SqliteConnectOptions::new() .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) .synchronous(SqliteSynchronous::Normal) diff --git a/nyx-chain-watcher/src/chain_scraper/mod.rs b/nyx-chain-watcher/src/chain_scraper/mod.rs index 61fbf0f6db..83205e5f03 100644 --- a/nyx-chain-watcher/src/chain_scraper/mod.rs +++ b/nyx-chain-watcher/src/chain_scraper/mod.rs @@ -1,9 +1,10 @@ use crate::env::vars::{NYXD_SCRAPER_START_HEIGHT, NYXD_SCRAPER_USE_BEST_EFFORT_START_HEIGHT}; use nyxd_scraper::{storage::ScraperStorage, NyxdScraper, PruningOptions}; +use tracing::info; pub(crate) async fn run_chain_scraper( config: &crate::config::Config, -) -> anyhow::Result { +) -> anyhow::Result { let websocket_url = std::env::var("NYXD_WS").expect("NYXD_WS not defined"); let rpc_url = std::env::var("NYXD").expect("NYXD not defined"); @@ -38,5 +39,8 @@ pub(crate) async fn run_chain_scraper( let instance = scraper.build_and_start().await?; - Ok(instance.storage) + info!("🚧 blocking until the chain has caught up..."); + instance.wait_for_startup_sync().await; + + Ok(instance) } diff --git a/nyx-chain-watcher/src/cli/commands/run/mod.rs b/nyx-chain-watcher/src/cli/commands/run/mod.rs index d3095fc37d..96344a4014 100644 --- a/nyx-chain-watcher/src/cli/commands/run/mod.rs +++ b/nyx-chain-watcher/src/cli/commands/run/mod.rs @@ -48,7 +48,8 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa // Spawn the payment listener task let payment_listener_handle = tokio::spawn({ let obs_pool = watcher_pool.clone(); - let chain_storage = run_chain_scraper(&config).await?; + let chain_scraper = run_chain_scraper(&config).await?; + let chain_storage = chain_scraper.storage(); let payment_watcher_config = config.payment_watcher_config.unwrap_or_default(); async move {