rename nyxd-scraper to sqlite
This commit is contained in:
Jędrzej Stuczyński
@@ -0,0 +1,65 @@
|
||||
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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<u32>) -> VecDeque<Range<u32>> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,515 @@
|
||||
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
|
||||
// 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;
|
||||
use crate::modules::{BlockModule, MsgModule, TxModule};
|
||||
use crate::rpc_client::RpcClient;
|
||||
use crate::storage::{persist_block, ScraperStorage};
|
||||
use crate::PruningOptions;
|
||||
use futures::StreamExt;
|
||||
use std::cmp::max;
|
||||
use std::collections::{BTreeMap, HashSet, VecDeque};
|
||||
use std::ops::{Add, Range};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc::{Sender, UnboundedReceiver};
|
||||
use tokio::sync::Notify;
|
||||
use tokio::time::{interval_at, Instant};
|
||||
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, instrument, trace, warn};
|
||||
|
||||
mod helpers;
|
||||
pub(crate) mod pruning;
|
||||
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<u32>,
|
||||
queued_requests: VecDeque<Range<u32>>,
|
||||
}
|
||||
|
||||
impl PendingSync {
|
||||
fn is_empty(&self) -> bool {
|
||||
self.request_in_flight.is_empty() && self.queued_requests.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct BlockProcessorConfig {
|
||||
pub pruning_options: PruningOptions,
|
||||
pub store_precommits: bool,
|
||||
pub explicit_starting_block_height: Option<u32>,
|
||||
pub use_best_effort_start_height: bool,
|
||||
}
|
||||
|
||||
impl Default for BlockProcessorConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
pruning_options: PruningOptions::nothing(),
|
||||
store_precommits: true,
|
||||
explicit_starting_block_height: None,
|
||||
use_best_effort_start_height: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl BlockProcessorConfig {
|
||||
pub fn new(
|
||||
pruning_options: PruningOptions,
|
||||
store_precommits: bool,
|
||||
explicit_starting_block_height: Option<u32>,
|
||||
use_best_effort_start_height: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
pruning_options,
|
||||
store_precommits,
|
||||
explicit_starting_block_height,
|
||||
use_best_effort_start_height,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct BlockProcessor {
|
||||
config: BlockProcessorConfig,
|
||||
cancel: CancellationToken,
|
||||
synced: Arc<Notify>,
|
||||
last_processed_height: u32,
|
||||
last_pruned_height: u32,
|
||||
last_processed_at: Instant,
|
||||
pending_sync: PendingSync,
|
||||
queued_blocks: BTreeMap<u32, BlockToProcess>,
|
||||
|
||||
rpc_client: RpcClient,
|
||||
incoming: UnboundedReceiverStream<BlockToProcess>,
|
||||
block_requester: Sender<BlockRequest>,
|
||||
storage: ScraperStorage,
|
||||
|
||||
// future work: rather than sending each msg to every msg module,
|
||||
// let them subscribe based on `type_url` inside the message itself
|
||||
// (like "/cosmwasm.wasm.v1.MsgExecuteContract")
|
||||
block_modules: Vec<Box<dyn BlockModule + Send>>,
|
||||
tx_modules: Vec<Box<dyn TxModule + Send>>,
|
||||
msg_modules: Vec<Box<dyn MsgModule + Send>>,
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
impl BlockProcessor {
|
||||
pub async fn new(
|
||||
config: BlockProcessorConfig,
|
||||
cancel: CancellationToken,
|
||||
synced: Arc<Notify>,
|
||||
incoming: UnboundedReceiver<BlockToProcess>,
|
||||
block_requester: Sender<BlockRequest>,
|
||||
storage: ScraperStorage,
|
||||
rpc_client: RpcClient,
|
||||
) -> Result<Self, ScraperError> {
|
||||
let last_processed = storage.get_last_processed_height().await?;
|
||||
let last_processed_height = last_processed.try_into().unwrap_or_default();
|
||||
|
||||
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,
|
||||
synced,
|
||||
last_processed_height,
|
||||
last_pruned_height,
|
||||
last_processed_at: Instant::now(),
|
||||
pending_sync: Default::default(),
|
||||
queued_blocks: Default::default(),
|
||||
rpc_client,
|
||||
incoming: incoming.into(),
|
||||
block_requester,
|
||||
storage,
|
||||
block_modules: vec![],
|
||||
tx_modules: vec![],
|
||||
msg_modules: vec![],
|
||||
})
|
||||
}
|
||||
|
||||
pub fn with_pruning(mut self, pruning_options: PruningOptions) -> Self {
|
||||
self.config.pruning_options = pruning_options;
|
||||
self
|
||||
}
|
||||
|
||||
pub(super) async fn process_block(
|
||||
&mut self,
|
||||
block: BlockToProcess,
|
||||
) -> Result<(), ScraperError> {
|
||||
info!("processing block at height {}", block.height);
|
||||
|
||||
let full_info = self.rpc_client.try_get_full_details(block).await?;
|
||||
|
||||
debug!(
|
||||
"this block has {} transaction(s)",
|
||||
full_info.transactions.len()
|
||||
);
|
||||
for tx in &full_info.transactions {
|
||||
debug!("{} has {} message(s)", tx.hash, tx.tx.body.messages.len());
|
||||
for (index, msg) in tx.tx.body.messages.iter().enumerate() {
|
||||
debug!("{index}: {:?}", msg.type_url)
|
||||
}
|
||||
}
|
||||
|
||||
// process the entire block as a transaction so that if anything fails,
|
||||
// we won't end up with a corrupted storage.
|
||||
let mut tx = self.storage.begin_processing_tx().await?;
|
||||
|
||||
persist_block(&full_info, &mut tx, self.config.store_precommits).await?;
|
||||
|
||||
// let the modules do whatever they want
|
||||
// the ones wanting the full block:
|
||||
for block_module in &mut self.block_modules {
|
||||
block_module.handle_block(&full_info, &mut tx).await?;
|
||||
}
|
||||
|
||||
// the ones wanting transactions:
|
||||
for block_tx in full_info.transactions {
|
||||
for tx_module in &mut self.tx_modules {
|
||||
tx_module.handle_tx(&block_tx, &mut tx).await?;
|
||||
}
|
||||
// the ones concerned with individual messages
|
||||
for (index, msg) in block_tx.tx.body.messages.iter().enumerate() {
|
||||
for msg_module in &mut self.msg_modules {
|
||||
if msg.type_url == msg_module.type_url() {
|
||||
msg_module
|
||||
.handle_msg(index, msg, &block_tx, &mut tx)
|
||||
.await?
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let commit_start = Instant::now();
|
||||
tx.commit()
|
||||
.await
|
||||
.map_err(|source| ScraperError::StorageTxCommitFailure { source })?;
|
||||
crate::storage::log_db_operation_time("committing processing tx", commit_start);
|
||||
|
||||
self.last_processed_height = full_info.block.header.height.value() as u32;
|
||||
self.last_processed_at = Instant::now();
|
||||
if let Err(err) = self.maybe_prune_storage().await {
|
||||
error!("failed to prune the storage: {err}");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn set_block_modules(&mut self, modules: Vec<Box<dyn BlockModule + Send>>) {
|
||||
self.block_modules = modules;
|
||||
}
|
||||
|
||||
pub fn set_tx_modules(&mut self, modules: Vec<Box<dyn TxModule + Send>>) {
|
||||
self.tx_modules = modules;
|
||||
}
|
||||
|
||||
pub fn set_msg_modules(&mut self, modules: Vec<Box<dyn MsgModule + Send>>) {
|
||||
self.msg_modules = modules;
|
||||
}
|
||||
|
||||
pub(super) fn last_process_height(&self) -> u32 {
|
||||
self.last_processed_height
|
||||
}
|
||||
|
||||
async fn maybe_request_missing_blocks(&mut self) -> Result<(), ScraperError> {
|
||||
// we're still processing, so we're good
|
||||
if self.last_processed_at.elapsed() < MAX_MISSING_BLOCKS_DELAY {
|
||||
debug!("no need to request missing blocks");
|
||||
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()
|
||||
{
|
||||
self.last_processed_height + 1..*next_available
|
||||
} else {
|
||||
let current_height = self.rpc_client.current_block_height().await? as u32;
|
||||
self.last_processed_height + 1..current_height + 1
|
||||
};
|
||||
|
||||
self.request_missing_blocks(request_range).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn request_missing_blocks(
|
||||
&mut self,
|
||||
request_range: Range<u32>,
|
||||
) -> Result<(), ScraperError> {
|
||||
let request_range = if request_range.len() > MAX_RANGE_SIZE {
|
||||
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<u32>) -> Result<(), ScraperError> {
|
||||
debug!("requesting missing blocks: {request_range:?}");
|
||||
|
||||
self.block_requester
|
||||
.send(BlockRequest::Range(request_range))
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
async fn prune_storage(&mut self) -> Result<(), ScraperError> {
|
||||
let keep_recent = self.config.pruning_options.strategy_keep_recent();
|
||||
let last_to_keep = self.last_processed_height - keep_recent;
|
||||
|
||||
info!(
|
||||
keep_recent,
|
||||
oldest_to_keep = last_to_keep,
|
||||
"pruning the storage"
|
||||
);
|
||||
|
||||
let lowest: u32 = self
|
||||
.storage
|
||||
.lowest_block_height()
|
||||
.await?
|
||||
.unwrap_or_default()
|
||||
.try_into()
|
||||
.unwrap_or_default();
|
||||
|
||||
let to_prune = last_to_keep.saturating_sub(lowest);
|
||||
match to_prune {
|
||||
v if v > 1000 => warn!("approximately {v} blocks worth of data will be pruned"),
|
||||
v if v > 100 => info!("approximately {v} blocks worth of data will be pruned"),
|
||||
0 => trace!("no blocks to prune"),
|
||||
v => debug!("approximately {v} blocks worth of data will be pruned"),
|
||||
}
|
||||
|
||||
if to_prune == 0 {
|
||||
self.last_pruned_height = self.last_processed_height;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
self.storage
|
||||
.prune_storage(last_to_keep, self.last_processed_height)
|
||||
.await?;
|
||||
|
||||
self.last_pruned_height = self.last_processed_height;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn maybe_prune_storage(&mut self) -> Result<(), ScraperError> {
|
||||
debug!("checking for storage pruning");
|
||||
|
||||
if self.config.pruning_options.strategy.is_nothing() {
|
||||
trace!("the current pruning strategy is 'nothing'");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let interval = self.config.pruning_options.strategy_interval();
|
||||
if self.last_pruned_height + interval <= self.last_processed_height {
|
||||
self.prune_storage().await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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");
|
||||
self.last_processed_height = height - 1
|
||||
}
|
||||
|
||||
if height <= self.last_processed_height {
|
||||
warn!("we have already processed block for height {height}");
|
||||
return;
|
||||
}
|
||||
|
||||
if self.last_processed_height + 1 != height {
|
||||
if self.queued_blocks.insert(height, block).is_some() {
|
||||
warn!("we have already queued up block for height {height}");
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if let Err(err) = self.process_block(block).await {
|
||||
error!("failed to process block at height {height}: {err}");
|
||||
return;
|
||||
}
|
||||
|
||||
// process as much as we can from the queue
|
||||
let mut next = height + 1;
|
||||
while let Some(next_block) = self.queued_blocks.remove(&next) {
|
||||
if let Err(err) = self.process_block(next_block).await {
|
||||
error!("failed to process queued-up block at height {next}: {err}")
|
||||
}
|
||||
next += 1;
|
||||
}
|
||||
|
||||
self.try_request_pending().await;
|
||||
|
||||
if self.pending_sync.is_empty() {
|
||||
self.synced.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
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());
|
||||
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();
|
||||
let last_to_keep = latest_block - keep_recent;
|
||||
|
||||
if !self.config.pruning_options.strategy.is_nothing() {
|
||||
self.last_processed_height = max(self.last_processed_height, last_to_keep);
|
||||
}
|
||||
|
||||
let request_range = self.last_processed_height + 1..latest_block + 1;
|
||||
info!(
|
||||
keep_recent = %keep_recent,
|
||||
last_to_keep = %last_to_keep,
|
||||
last_processed_height = %self.last_processed_height,
|
||||
"we need to request {request_range:?} to resync"
|
||||
);
|
||||
self.request_missing_blocks(request_range).await?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// 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(());
|
||||
};
|
||||
|
||||
info!("attempting to start the scraper from block {starting_height}");
|
||||
let earliest_available =
|
||||
self.rpc_client.earliest_available_block_height().await? as u32;
|
||||
info!("earliest available block height: {earliest_available}");
|
||||
|
||||
if earliest_available > starting_height && self.config.use_best_effort_start_height {
|
||||
error!("the earliest available block is higher than the desired starting height");
|
||||
return Err(ScraperError::BlocksUnavailable {
|
||||
height: starting_height,
|
||||
});
|
||||
}
|
||||
|
||||
let starting_height = if earliest_available > starting_height {
|
||||
// add few additional blocks to account for all the startup waiting
|
||||
// because the node might have pruned few blocks since
|
||||
earliest_available + 10
|
||||
} else {
|
||||
starting_height
|
||||
};
|
||||
|
||||
let request_range = starting_height..latest_block + 1;
|
||||
|
||||
info!("going to start the scraper from block {starting_height}");
|
||||
info!("we need to request {request_range:?} before properly starting up");
|
||||
|
||||
self.request_missing_blocks(request_range).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn run(&mut self) {
|
||||
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
|
||||
let mut missing_check_interval = interval_at(
|
||||
Instant::now().add(MISSING_BLOCKS_CHECK_INTERVAL),
|
||||
MISSING_BLOCKS_CHECK_INTERVAL,
|
||||
);
|
||||
|
||||
if let Err(err) = self.startup_resync().await {
|
||||
error!("failed to perform startup sync: {err}");
|
||||
self.cancel.cancel();
|
||||
return;
|
||||
};
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = self.cancel.cancelled() => {
|
||||
info!("received cancellation token");
|
||||
break
|
||||
}
|
||||
_ = missing_check_interval.tick() => {
|
||||
if let Err(err) = self.maybe_request_missing_blocks().await {
|
||||
error!("failed to request missing blocks: {err}")
|
||||
}
|
||||
}
|
||||
block = self.incoming.next() => {
|
||||
match block {
|
||||
Some(block) => self.next_incoming(block).await,
|
||||
None => {
|
||||
warn!("stopped receiving new blocks");
|
||||
self.cancel.cancel();
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use crate::error::ScraperError;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
pub const DEFAULT_PRUNING_KEEP_RECENT: u32 = 362880;
|
||||
pub const DEFAULT_PRUNING_INTERVAL: u32 = 10;
|
||||
pub const EVERYTHING_PRUNING_KEEP_RECENT: u32 = 2;
|
||||
pub const EVERYTHING_PRUNING_INTERVAL: u32 = 10;
|
||||
|
||||
/// We follow cosmos-sdk pruning strategies for convenience’s sake.
|
||||
#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum PruningStrategy {
|
||||
/// 'Default' strategy defines a pruning strategy where the last 362880 heights are
|
||||
/// kept where to-be pruned heights are pruned at every 10th height.
|
||||
/// The last 362880 heights are kept(approximately 3.5 weeks worth of state) assuming the typical
|
||||
/// block time is 6s. If these values do not match the applications' requirements, use the "custom" option.
|
||||
#[default]
|
||||
Default,
|
||||
|
||||
/// 'Everything' strategy defines a pruning strategy where all committed heights are
|
||||
/// deleted, storing only the current height and last 2 states. To-be pruned heights are
|
||||
/// pruned at every 10th height.
|
||||
Everything,
|
||||
|
||||
/// 'Nothing' strategy defines a pruning strategy where all heights are kept on disk.
|
||||
Nothing,
|
||||
|
||||
/// 'Custom' strategy defines a pruning strategy where the user specifies the pruning.
|
||||
Custom,
|
||||
}
|
||||
|
||||
impl PruningStrategy {
|
||||
pub fn is_custom(&self) -> bool {
|
||||
matches!(self, PruningStrategy::Custom)
|
||||
}
|
||||
|
||||
pub fn is_nothing(&self) -> bool {
|
||||
matches!(self, PruningStrategy::Nothing)
|
||||
}
|
||||
|
||||
pub fn is_everything(&self) -> bool {
|
||||
matches!(self, PruningStrategy::Everything)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
|
||||
pub struct PruningOptions {
|
||||
/// keep_recent defines how many recent heights to keep on disk.
|
||||
pub keep_recent: u32,
|
||||
|
||||
/// interval defines the frequency of removing the pruned heights from the disk.
|
||||
pub interval: u32,
|
||||
|
||||
/// strategy defines the currently used kind of [PruningStrategy].
|
||||
pub strategy: PruningStrategy,
|
||||
}
|
||||
|
||||
impl PruningOptions {
|
||||
pub fn validate(&self) -> Result<(), ScraperError> {
|
||||
// if strategy is not set to custom, other options are meaningless since they won't be applied
|
||||
if !self.strategy.is_custom() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if self.interval == 0 {
|
||||
return Err(ScraperError::ZeroPruningInterval);
|
||||
}
|
||||
|
||||
if self.interval < EVERYTHING_PRUNING_INTERVAL {
|
||||
return Err(ScraperError::TooSmallPruningInterval {
|
||||
interval: self.interval,
|
||||
});
|
||||
}
|
||||
|
||||
if self.keep_recent < EVERYTHING_PRUNING_KEEP_RECENT {
|
||||
return Err(ScraperError::TooSmallKeepRecent {
|
||||
keep_recent: self.keep_recent,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn nothing() -> Self {
|
||||
PruningOptions {
|
||||
keep_recent: 0,
|
||||
interval: 0,
|
||||
strategy: PruningStrategy::Nothing,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn strategy_interval(&self) -> u32 {
|
||||
match self.strategy {
|
||||
PruningStrategy::Default => DEFAULT_PRUNING_INTERVAL,
|
||||
PruningStrategy::Everything => EVERYTHING_PRUNING_INTERVAL,
|
||||
PruningStrategy::Nothing => 0,
|
||||
PruningStrategy::Custom => self.interval,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn strategy_keep_recent(&self) -> u32 {
|
||||
match self.strategy {
|
||||
PruningStrategy::Default => DEFAULT_PRUNING_KEEP_RECENT,
|
||||
PruningStrategy::Everything => EVERYTHING_PRUNING_KEEP_RECENT,
|
||||
PruningStrategy::Nothing => 0,
|
||||
PruningStrategy::Custom => self.keep_recent,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for PruningOptions {
|
||||
fn default() -> Self {
|
||||
PruningOptions {
|
||||
keep_recent: DEFAULT_PRUNING_KEEP_RECENT,
|
||||
interval: DEFAULT_PRUNING_INTERVAL,
|
||||
strategy: Default::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
use crate::error::ScraperError;
|
||||
use crate::helpers;
|
||||
use tendermint::{abci, block, tx, Block, Hash};
|
||||
use tendermint_rpc::endpoint::{block as block_endpoint, block_results, validators};
|
||||
use tendermint_rpc::event::{Event, EventData};
|
||||
|
||||
// just get all everything out of tx::Response, but parse raw `tx` bytes
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ParsedTransactionResponse {
|
||||
/// The hash of the transaction.
|
||||
///
|
||||
/// Deserialized from a hex-encoded string (there is a discrepancy between
|
||||
/// the format used for the request and the format used for the response in
|
||||
/// the Tendermint RPC).
|
||||
pub hash: Hash,
|
||||
|
||||
pub height: block::Height,
|
||||
|
||||
pub index: u32,
|
||||
|
||||
pub tx_result: abci::types::ExecTxResult,
|
||||
|
||||
pub tx: cosmrs::tx::Tx,
|
||||
|
||||
pub proof: Option<tx::Proof>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct FullBlockInformation {
|
||||
/// Basic block information, including its signers.
|
||||
pub block: Block,
|
||||
|
||||
/// All of the emitted events alongside any tx results.
|
||||
pub results: block_results::Response,
|
||||
|
||||
/// Validator set for this particular block
|
||||
pub validators: validators::Response,
|
||||
|
||||
/// Transaction results from this particular block
|
||||
pub transactions: Vec<ParsedTransactionResponse>,
|
||||
}
|
||||
|
||||
impl FullBlockInformation {
|
||||
pub fn ensure_proposer(&self) -> Result<(), ScraperError> {
|
||||
let block_proposer = self.block.header.proposer_address;
|
||||
if !self
|
||||
.validators
|
||||
.validators
|
||||
.iter()
|
||||
.any(|v| v.address == block_proposer)
|
||||
{
|
||||
let proposer = helpers::validator_consensus_address(block_proposer)?;
|
||||
return Err(ScraperError::BlockProposerNotInValidatorSet {
|
||||
height: self.block.header.height.value() as u32,
|
||||
proposer: proposer.to_string(),
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct BlockToProcess {
|
||||
pub(crate) height: u32,
|
||||
pub(crate) block: Block,
|
||||
}
|
||||
|
||||
impl From<Block> for BlockToProcess {
|
||||
fn from(block: Block) -> Self {
|
||||
BlockToProcess {
|
||||
height: block.header.height.value() as u32,
|
||||
block,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<Event> for BlockToProcess {
|
||||
type Error = ScraperError;
|
||||
|
||||
fn try_from(event: Event) -> Result<Self, Self::Error> {
|
||||
let query = event.query.clone();
|
||||
|
||||
// TODO: we're losing `result_begin_block` and `result_end_block` here but maybe that's fine?
|
||||
let maybe_block = match event.data {
|
||||
EventData::NewBlock { block, .. } => block,
|
||||
EventData::LegacyNewBlock { block, .. } => block,
|
||||
EventData::Tx { .. } => {
|
||||
return Err(ScraperError::InvalidSubscriptionEvent {
|
||||
query,
|
||||
kind: "Tx".to_string(),
|
||||
})
|
||||
}
|
||||
EventData::GenericJsonEvent(_) => {
|
||||
return Err(ScraperError::InvalidSubscriptionEvent {
|
||||
query,
|
||||
kind: "GenericJsonEvent".to_string(),
|
||||
})
|
||||
}
|
||||
};
|
||||
|
||||
let Some(block) = maybe_block else {
|
||||
return Err(ScraperError::EmptyBlockData { query });
|
||||
};
|
||||
|
||||
Ok((*block).into())
|
||||
}
|
||||
}
|
||||
|
||||
impl From<block_endpoint::Response> for BlockToProcess {
|
||||
fn from(value: block_endpoint::Response) -> Self {
|
||||
value.block.into()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user