From 50cc8bd0bf2c5139a9006da608bb261dd1e6ed2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 13 Dec 2023 11:03:41 +0000 Subject: [PATCH] improved startup sync --- .../src/block_processor/helpers.rs | 65 +++++++++++++++ .../nyxd-scraper/src/block_processor/mod.rs | 79 ++++++++++++++++--- common/nyxd-scraper/src/storage/mod.rs | 2 +- 3 files changed, 136 insertions(+), 10 deletions(-) create mode 100644 common/nyxd-scraper/src/block_processor/helpers.rs diff --git a/common/nyxd-scraper/src/block_processor/helpers.rs b/common/nyxd-scraper/src/block_processor/helpers.rs new file mode 100644 index 0000000000..ffc8c592c5 --- /dev/null +++ b/common/nyxd-scraper/src/block_processor/helpers.rs @@ -0,0 +1,65 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::block_processor::MAX_RANGE_SIZE; +use std::cmp::min; +use std::collections::VecDeque; +use std::ops::Range; + +pub(crate) fn split_request_range(request_range: Range) -> VecDeque> { + let mut requests = VecDeque::new(); + + let mut start = request_range.start; + let mut end = min(request_range.end, start + MAX_RANGE_SIZE as u32); + + loop { + requests.push_back(start..end); + start = min(start + MAX_RANGE_SIZE as u32, request_range.end); + end = min(end + MAX_RANGE_SIZE as u32, request_range.end); + + if start == end { + break; + } + } + + requests +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn splitting_request_range() { + let range = 0..100; + let mut expected = VecDeque::new(); + expected.push_back(0..30); + expected.push_back(30..60); + expected.push_back(60..90); + expected.push_back(90..100); + assert_eq!(expected, split_request_range(range)); + + let range = 0..30; + let mut expected = VecDeque::new(); + expected.push_back(0..30); + assert_eq!(expected, split_request_range(range)); + + let range = 0..60; + let mut expected = VecDeque::new(); + expected.push_back(0..30); + expected.push_back(30..60); + assert_eq!(expected, split_request_range(range)); + + let range = 0..5; + let mut expected = VecDeque::new(); + expected.push_back(0..5); + assert_eq!(expected, split_request_range(range)); + + let range = 123..200; + let mut expected = VecDeque::new(); + expected.push_back(123..153); + expected.push_back(153..183); + expected.push_back(183..200); + assert_eq!(expected, split_request_range(range)); + } +} diff --git a/common/nyxd-scraper/src/block_processor/mod.rs b/common/nyxd-scraper/src/block_processor/mod.rs index 89e51d06b1..bb3e886b0e 100644 --- a/common/nyxd-scraper/src/block_processor/mod.rs +++ b/common/nyxd-scraper/src/block_processor/mod.rs @@ -1,6 +1,7 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +use crate::block_processor::helpers::split_request_range; use crate::block_processor::types::BlockToProcess; use crate::block_requester::BlockRequest; use crate::error::ScraperError; @@ -8,7 +9,7 @@ use crate::modules::{BlockModule, MsgModule, TxModule}; use crate::rpc_client::RpcClient; use crate::storage::{persist_block, ScraperStorage}; use futures::StreamExt; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet, VecDeque}; use std::ops::{Add, Range}; use std::time::Duration; use tokio::sync::mpsc::{Sender, UnboundedReceiver}; @@ -17,16 +18,30 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; +mod helpers; pub(crate) mod types; const MISSING_BLOCKS_CHECK_INTERVAL: Duration = Duration::from_secs(30); const MAX_MISSING_BLOCKS_DELAY: Duration = Duration::from_secs(15); const MAX_RANGE_SIZE: usize = 30; +#[derive(Debug, Default)] +struct PendingSync { + request_in_flight: HashSet, + queued_requests: VecDeque>, +} + +impl PendingSync { + fn is_empty(&self) -> bool { + self.request_in_flight.is_empty() && self.queued_requests.is_empty() + } +} + pub struct BlockProcessor { cancel: CancellationToken, last_processed_height: u32, last_processed_at: Instant, + pending_sync: PendingSync, queued_blocks: BTreeMap, rpc_client: RpcClient, @@ -56,6 +71,7 @@ impl BlockProcessor { cancel, last_processed_height: last_processed.try_into().unwrap_or_default(), last_processed_at: Instant::now(), + pending_sync: Default::default(), queued_blocks: Default::default(), rpc_client, incoming: incoming.into(), @@ -139,6 +155,10 @@ impl BlockProcessor { return Ok(()); } + if self.try_request_pending().await { + return Ok(()); + } + // TODO: properly fill in the gaps later with BlockRequest::Specific, let request_range = if let Some((next_available, _)) = self.queued_blocks.first_key_value() { @@ -148,22 +168,37 @@ impl BlockProcessor { self.last_processed_height + 1..current_height + 1 }; - self.request_block_range(request_range).await?; + self.request_missing_blocks(request_range).await?; Ok(()) } - // technically we're not mutating self here, - // but we need it to help the compiler figure out the future is `Send` - async fn request_block_range(&mut self, request_range: Range) -> Result<(), ScraperError> { - debug!("requesting missing blocks: {request_range:?}"); + async fn request_missing_blocks( + &mut self, + request_range: Range, + ) -> Result<(), ScraperError> { let request_range = if request_range.len() > MAX_RANGE_SIZE { - warn!("the request range has more than {MAX_RANGE_SIZE} values. reducing it."); - request_range.start..request_range.start + MAX_RANGE_SIZE as u32 + let mut split = split_request_range(request_range); + + // SAFETY: we know that after the split of a non-empty range we have AT LEAST one value + #[allow(clippy::unwrap_used)] + let first = split.pop_front().unwrap(); + self.pending_sync.queued_requests = split; + self.pending_sync.request_in_flight = first.clone().collect(); + + first } else { request_range }; + self.send_blocks_request(request_range).await + } + + // technically we're not mutating self here, + // but we need it to help the compiler figure out the future is `Send` + async fn send_blocks_request(&mut self, request_range: Range) -> Result<(), ScraperError> { + debug!("requesting missing blocks: {request_range:?}"); + self.block_requester .send(BlockRequest::Range(request_range)) .await?; @@ -173,6 +208,8 @@ impl BlockProcessor { async fn next_incoming(&mut self, block: BlockToProcess) { let height = block.height; + self.pending_sync.request_in_flight.remove(&height); + if self.last_processed_height == 0 { // this is the first time we've started up the process debug!("setting up initial processing height"); @@ -204,16 +241,40 @@ impl BlockProcessor { } next += 1; } + + self.try_request_pending().await; + } + + async fn try_request_pending(&mut self) -> bool { + if self.pending_sync.request_in_flight.is_empty() { + if let Some(next_sync) = self.pending_sync.queued_requests.pop_front() { + debug!( + "current request range has been resolved. requesting another bunch of blocks" + ); + if let Err(err) = self.send_blocks_request(next_sync.clone()).await { + error!("failed to request resync blocks: {err}"); + self.pending_sync.queued_requests.push_front(next_sync); + } else { + self.pending_sync.request_in_flight = next_sync.collect() + } + + return true; + } + } + + false } // technically we're not mutating self here, // 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()); + let latest_block = self.rpc_client.current_block_height().await? as u32; if latest_block > self.last_processed_height && self.last_processed_height != 0 { let request_range = self.last_processed_height + 1..latest_block + 1; info!("we need to request {request_range:?} to resync"); - self.request_block_range(request_range).await?; + self.request_missing_blocks(request_range).await?; } Ok(()) diff --git a/common/nyxd-scraper/src/storage/mod.rs b/common/nyxd-scraper/src/storage/mod.rs index ce491e5663..1889752a97 100644 --- a/common/nyxd-scraper/src/storage/mod.rs +++ b/common/nyxd-scraper/src/storage/mod.rs @@ -69,7 +69,7 @@ impl ScraperStorage { #[instrument(skip_all)] pub async fn begin_processing_tx(&self) -> Result { - info!("starting storage tx"); + debug!("starting storage tx"); self.manager .connection_pool .begin()