From 681c886f8b0dce091007e40280ace14e18d8cfca Mon Sep 17 00:00:00 2001 From: gventino-cw Date: Wed, 19 Aug 2026 15:16:27 -0300 Subject: [PATCH 1/3] feat: generic pagination engine with cursor paginatio policy + importer pagination --- .../follower/importer/importer_supervisor.rs | 5 +- .../blockchain_client/blockchain_client.rs | 27 +- .../blockchain_client/importer_pagination.rs | 249 ++++++++++++++++++ src/eth/rpc/blockchain_client/mod.rs | 1 + src/eth/rpc/server.rs | 29 +- src/eth/rpc/types/importer_pagination.rs | 200 ++++++++++++++ src/eth/rpc/types/mod.rs | 11 + src/eth/rpc/types/pagination.rs | 220 ++++++++++++++++ src/eth/types/external/external_block.rs | 32 +++ 9 files changed, 747 insertions(+), 27 deletions(-) create mode 100644 src/eth/rpc/blockchain_client/importer_pagination.rs create mode 100644 src/eth/rpc/types/importer_pagination.rs create mode 100644 src/eth/rpc/types/pagination.rs diff --git a/src/eth/follower/importer/importer_supervisor.rs b/src/eth/follower/importer/importer_supervisor.rs index 44046529e..b680dc8a7 100644 --- a/src/eth/follower/importer/importer_supervisor.rs +++ b/src/eth/follower/importer/importer_supervisor.rs @@ -141,10 +141,11 @@ pub async fn start_importer( .run(resume_from, sync_interval, chain, stop_at_block) .await?; } - ImporterMode::FakeLeader => + ImporterMode::FakeLeader => { FakeLeader::new(executor, miner, storage, Arc::clone(&chain)) .run(resume_from, sync_interval, chain, stop_at_block) - .await?, + .await?; + } } Ok(()) } diff --git a/src/eth/rpc/blockchain_client/blockchain_client.rs b/src/eth/rpc/blockchain_client/blockchain_client.rs index 9f4aa6854..5074973ae 100644 --- a/src/eth/rpc/blockchain_client/blockchain_client.rs +++ b/src/eth/rpc/blockchain_client/blockchain_client.rs @@ -13,6 +13,7 @@ use jsonrpsee::ws_client::WsClientBuilder; use tokio::sync::RwLock; use tokio::sync::RwLockReadGuard; +use super::importer_pagination::ImporterPaginationClient; use crate::GlobalState; use crate::alias::AlloyBytes; use crate::alias::AlloyTransaction; @@ -36,7 +37,7 @@ use crate::log_and_err; #[derive(Debug)] pub struct BlockchainClient { - http: HttpClient, + pub(super) http: HttpClient, pub http_url: String, ws: Option>, ws_url: Option, @@ -159,33 +160,13 @@ impl BlockchainClient { /// Fetches a block by number with receipts. pub async fn fetch_block_and_receipts(&self, block_number: BlockNumber) -> anyhow::Result> { tracing::debug!(%block_number, "fetching block"); - - let number = to_json_value(block_number); - let result = self - .http - .request::, _>("stratus_getBlockAndReceipts", [number]) - .await; - - match result { - Ok(block) => Ok(block), - Err(e) => log_and_err!(reason = e, "failed to fetch block with receipts"), - } + ImporterPaginationClient::new(self).fetch_block_and_receipts(block_number).await } /// Fetches a block by number with changes. pub async fn fetch_block_with_changes(&self, block_number: BlockNumber) -> anyhow::Result> { tracing::debug!(%block_number, "fetching block with changes"); - - let number = to_json_value(block_number); - let result = self - .http - .request::, _>("stratus_getBlockWithChanges", [number]) - .await; - - match result { - Ok(block) => Ok(block), - Err(e) => log_and_err!(reason = e, "failed to fetch block with changes"), - } + ImporterPaginationClient::new(self).fetch_block_with_changes(block_number).await } /// Fetches a block by number. diff --git a/src/eth/rpc/blockchain_client/importer_pagination.rs b/src/eth/rpc/blockchain_client/importer_pagination.rs new file mode 100644 index 000000000..70fff7029 --- /dev/null +++ b/src/eth/rpc/blockchain_client/importer_pagination.rs @@ -0,0 +1,249 @@ +use anyhow::bail; +use jsonrpsee::core::client::ClientT; +use serde::de::DeserializeOwned; + +use super::blockchain_client::BlockchainClient; +use crate::eth::rpc::types::BlockAndReceiptsPageResponse; +use crate::eth::rpc::types::BlockWithChangesPageResponse; +use crate::eth::rpc::types::IMPORTER_PAGE_LIMIT_DEFAULT; +use crate::eth::rpc::types::ImporterCursorPaginator; +use crate::eth::rpc::types::ImporterPageInfo; +use crate::eth::rpc::types::ImporterPageRequest; +use crate::eth::rpc::types::PageReducer; +use crate::eth::rpc::types::PaginatedPageFetcher; +use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; +use crate::eth::storage::permanent::rocks::types::BlockRocksdb; +use crate::eth::types::BlockNumber; +use crate::eth::types::ExternalBlock; +use crate::eth::types::ExternalBlockWithReceipts; +use crate::eth::types::ExternalReceipt; +use crate::ext::to_json_value; +use crate::log_and_err; + +const GET_BLOCK_AND_RECEIPTS: &str = "stratus_getBlockAndReceipts"; +const GET_BLOCK_WITH_CHANGES: &str = "stratus_getBlockWithChanges"; + +pub(super) struct ImporterPaginationClient<'a> { + client: &'a BlockchainClient, +} + +impl<'a> ImporterPaginationClient<'a> { + pub(super) fn new(client: &'a BlockchainClient) -> Self { + Self { client } + } + + pub(super) async fn fetch_block_and_receipts(&self, block_number: BlockNumber) -> anyhow::Result> { + PaginatedPageFetcher::new(BlockAndReceiptsPages::new(block_number)) + .collect(|cursor| self.fetch_page(GET_BLOCK_AND_RECEIPTS, block_number, cursor, "failed to fetch block with receipts")) + .await + } + + pub(super) async fn fetch_block_with_changes(&self, block_number: BlockNumber) -> anyhow::Result> { + PaginatedPageFetcher::new(BlockWithChangesPages::new(block_number)) + .collect(|cursor| self.fetch_page(GET_BLOCK_WITH_CHANGES, block_number, cursor, "failed to fetch block with changes")) + .await + } + + async fn fetch_page(&self, method: &str, block_number: BlockNumber, cursor: Option, error_message: &'static str) -> anyhow::Result> + where + T: DeserializeOwned, + { + let params = [ + to_json_value(block_number), + to_json_value(ImporterPageRequest { + cursor, + limit: Some(IMPORTER_PAGE_LIMIT_DEFAULT), + }), + ]; + + match self.client.http.request::, _>(method, params).await { + Ok(page) => Ok(page), + Err(e) => log_and_err!(reason = e, error_message), + } + } +} + +fn validate_progress(page: &ImporterPageInfo, expected_total: &mut Option, context: &str) -> anyhow::Result> { + if page.returned == 0 && page.next_cursor.is_some() { + bail!("paginated {context} returned no items but provided a next cursor"); + } + + match expected_total { + Some(expected_total) if *expected_total != page.total => { + bail!("paginated {context} changed total from {expected_total} to {}", page.total); + } + Some(_) => {} + None => *expected_total = Some(page.total), + } + + Ok(page.next_cursor.clone()) +} + +struct BlockAndReceiptsPages { + block_number: BlockNumber, + block: Option, + receipts: Vec, + expected_total: Option, +} + +impl BlockAndReceiptsPages { + fn new(block_number: BlockNumber) -> Self { + Self { + block_number, + block: None, + receipts: Vec::new(), + expected_total: None, + } + } + + fn push_block(&mut self, page_block: ExternalBlock) -> anyhow::Result<()> { + if page_block.number() != self.block_number { + bail!( + "paginated block with receipts returned unexpected block number {} instead of {}", + page_block.number(), + self.block_number + ); + } + + match &mut self.block { + Some(block) => block.extend_full_transactions_from(page_block), + None => { + self.block = Some(page_block); + Ok(()) + } + } + } +} + +impl PageReducer for BlockAndReceiptsPages { + type Output = ExternalBlockWithReceipts; + type Paginator = ImporterCursorPaginator; + + fn reduce(&mut self, page: BlockAndReceiptsPageResponse) -> anyhow::Result> { + let cursor = validate_progress(&page.pagination, &mut self.expected_total, "block with receipts")?; + let page_block = ExternalBlock::try_from(page.block)?; + self.push_block(page_block)?; + self.receipts.extend(page.receipts); + Ok(cursor) + } + + fn finish_after_not_found(self) -> anyhow::Result> { + if self.block.is_none() { + Ok(None) + } else { + bail!("block disappeared while fetching paginated block with receipts"); + } + } + + fn finish(self) -> anyhow::Result> { + let Some(block) = self.block else { + return Ok(None); + }; + + let expected_total = self.expected_total.unwrap_or_default(); + let transactions_len = block.full_transactions_len()?; + if transactions_len != expected_total { + bail!("paginated block with receipts assembled {transactions_len} transactions but expected {expected_total}"); + } + if transactions_len != self.receipts.len() { + bail!( + "paginated block with receipts assembled {} transactions but {} receipts", + transactions_len, + self.receipts.len() + ); + } + + Ok(Some(ExternalBlockWithReceipts { + block, + receipts: self.receipts, + })) + } +} + +struct BlockWithChangesPages { + block_number: BlockNumber, + block: Option, + changes: BlockChangesRocksdb, + expected_total: Option, +} + +impl BlockWithChangesPages { + fn new(block_number: BlockNumber) -> Self { + Self { + block_number, + block: None, + changes: BlockChangesRocksdb::default(), + expected_total: None, + } + } + + fn push_block(&mut self, page_block: BlockRocksdb) -> anyhow::Result<()> { + let page_block_number = BlockNumber::from(page_block.header.number); + if page_block_number != self.block_number { + bail!( + "paginated block with changes returned unexpected block number {page_block_number} instead of {}", + self.block_number + ); + } + + match &mut self.block { + Some(block) => { + if block.header.hash != page_block.header.hash { + bail!("paginated block with changes changed block hash"); + } + block.transactions.extend(page_block.transactions); + } + None => self.block = Some(page_block), + } + + Ok(()) + } + + fn push_changes(&mut self, page_changes: BlockChangesRocksdb) -> anyhow::Result<()> { + for (address, change) in page_changes.account_changes { + if self.changes.account_changes.insert(address, change).is_some() { + bail!("paginated block with changes returned duplicate account change"); + } + } + for (slot, value) in page_changes.slot_changes { + if self.changes.slot_changes.insert(slot, value).is_some() { + bail!("paginated block with changes returned duplicate slot change"); + } + } + Ok(()) + } +} + +impl PageReducer for BlockWithChangesPages { + type Output = (BlockRocksdb, BlockChangesRocksdb); + type Paginator = ImporterCursorPaginator; + + fn reduce(&mut self, page: BlockWithChangesPageResponse) -> anyhow::Result> { + let cursor = validate_progress(&page.pagination, &mut self.expected_total, "block with changes")?; + self.push_block(page.block)?; + self.push_changes(page.changes)?; + Ok(cursor) + } + + fn finish_after_not_found(self) -> anyhow::Result> { + if self.block.is_none() { + Ok(None) + } else { + bail!("block disappeared while fetching paginated block with changes"); + } + } + + fn finish(self) -> anyhow::Result> { + let Some(block) = self.block else { + return Ok(None); + }; + + let expected_total = self.expected_total.unwrap_or_default(); + let total = block.transactions.len() + self.changes.account_changes.len() + self.changes.slot_changes.len(); + if total != expected_total { + bail!("paginated block with changes assembled {total} items but expected {expected_total}"); + } + + Ok(Some((block, self.changes))) + } +} diff --git a/src/eth/rpc/blockchain_client/mod.rs b/src/eth/rpc/blockchain_client/mod.rs index 721ae25ac..566e98a09 100644 --- a/src/eth/rpc/blockchain_client/mod.rs +++ b/src/eth/rpc/blockchain_client/mod.rs @@ -1,4 +1,5 @@ #[allow(clippy::module_inception)] pub mod blockchain_client; +mod importer_pagination; pub use blockchain_client::BlockchainClient; diff --git a/src/eth/rpc/server.rs b/src/eth/rpc/server.rs index 5c5d476a3..bfd06a879 100644 --- a/src/eth/rpc/server.rs +++ b/src/eth/rpc/server.rs @@ -72,6 +72,7 @@ use crate::eth::rpc::next_rpc_param; use crate::eth::rpc::next_rpc_param_or_default; use crate::eth::rpc::parser::RpcExtensionsExt; use crate::eth::rpc::subscriptions::RpcSubscriptionsHandles; +use crate::eth::rpc::types::ImporterPagination; use crate::eth::storage::ExecutionKind; use crate::eth::storage::StorageError; use crate::eth::storage::StratusStorage; @@ -897,11 +898,23 @@ fn stratus_get_block_and_receipts(params: Params<'_>, ctx: Arc, ext: let _method_enter = info_span!("rpc::stratus_getBlockAndReceipts").entered(); // parse params - let (_, filter) = next_rpc_param::(params.sequence())?; + let (params, filter) = next_rpc_param::(params.sequence())?; + let pagination = ImporterPagination::next(params, filter)?; // track tracing::info!(%filter, "reading block and receipts"); + if let Some((filter, pagination)) = pagination { + let Some(block) = ctx.server.storage.read_block(filter)? else { + tracing::info!(%filter, "block not found"); + return Ok(JsonValue::Null); + }; + + let response = pagination.block_and_receipts_response(block)?; + tracing::info!(%filter, returned = response.pagination.returned, total = response.pagination.total, "block with transactions page found"); + return Ok(json!(response)); + } + let Some(block) = ctx.server.storage.read_block(filter)? else { tracing::info!(%filter, "block not found"); return Ok(JsonValue::Null); @@ -922,11 +935,23 @@ fn stratus_get_block_with_changes(params: Params<'_>, ctx: Arc, ext: let _method_enter = info_span!("rpc::stratus_getBlockWithChanges").entered(); // parse params - let (_, filter) = next_rpc_param::(params.sequence())?; + let (params, filter) = next_rpc_param::(params.sequence())?; + let pagination = ImporterPagination::next(params, filter)?; // track tracing::info!(%filter, "reading block and changes"); + if let Some((filter, pagination)) = pagination { + let Some((block, changes)) = ctx.server.storage.read_block_with_changes(filter)? else { + tracing::info!(%filter, "block not found"); + return Ok(JsonValue::Null); + }; + + let response = pagination.block_with_changes_response(block, changes)?; + tracing::info!(%filter, returned = response.pagination.returned, total = response.pagination.total, "block with changes page found"); + return Ok(json!(response)); + } + let Some(block) = ctx.server.storage.read_block_with_changes(filter)? else { tracing::info!(%filter, "block not found"); return Ok(JsonValue::Null); diff --git a/src/eth/rpc/types/importer_pagination.rs b/src/eth/rpc/types/importer_pagination.rs new file mode 100644 index 000000000..77bb69a6e --- /dev/null +++ b/src/eth/rpc/types/importer_pagination.rs @@ -0,0 +1,200 @@ +use jsonrpsee::types::ParamsSequence; + +use super::BlockFilter; +use super::RpcError; +use super::pagination::CursorCodec; +use super::pagination::CursorPageInfo; +use super::pagination::CursorPaginationPolicy; +use super::pagination::CursorPaginator; +use super::pagination::Paginator; +use crate::alias::AlloyReceipt; +use crate::alias::JsonValue; +use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb; +use crate::eth::storage::permanent::rocks::types::AddressRocksdb; +use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; +use crate::eth::storage::permanent::rocks::types::BlockRocksdb; +use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb; +use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb; +use crate::eth::types::Block; +use crate::eth::types::ExternalReceipt; +use crate::eth::types::Hash; + +pub const IMPORTER_PAGE_LIMIT_DEFAULT: usize = 256; +pub const IMPORTER_PAGE_LIMIT_MAX: usize = 5_000; + +const IMPORTER_CURSOR_VERSION: &str = "v1"; + +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ImporterPageRequest { + pub cursor: Option, + pub limit: Option, +} + +impl ImporterPageRequest { + fn parse_next(mut params: ParamsSequence<'_>) -> Result, RpcError> { + match params.optional_next::() { + Ok(page_request) => Ok(page_request), + Err(e) => Err(RpcError::ParameterDecodeError { + rust_type: "ImporterPageRequest", + decode_error: e.data().map(|x| x.to_string()).unwrap_or_default(), + }), + } + } + + fn limit(&self) -> usize { + match self.limit { + Some(0) | None => IMPORTER_PAGE_LIMIT_DEFAULT, + Some(limit) => limit.min(IMPORTER_PAGE_LIMIT_MAX), + } + } +} + +pub struct ImporterPagination { + request: ImporterPageRequest, + start: usize, +} + +impl ImporterPagination { + pub fn next(params: ParamsSequence<'_>, filter: BlockFilter) -> Result, RpcError> { + let Some(request) = ImporterPageRequest::parse_next(params)? else { + return Ok(None); + }; + let (filter, start) = Self::resolve_filter(filter, request.cursor.as_deref())?; + Ok(Some((filter, Self { request, start }))) + } + + pub fn block_and_receipts_response(&self, block: Block) -> Result { + let mut block = block; + let mut paginator = self.cursor_paginator(block.transactions.len(), block.hash())?; + let tx_range = paginator.take(block.transactions.len()); + let transactions = block.transactions[tx_range].to_vec(); + let receipts = transactions.iter().cloned().map(AlloyReceipt::from).map(ExternalReceipt).collect::>(); + + block.transactions = transactions; + + Ok(BlockAndReceiptsPageResponse { + block: block.to_json_rpc_with_full_transactions(), + receipts, + pagination: paginator.finish(), + }) + } + + pub fn block_with_changes_response(&self, block: BlockRocksdb, changes: BlockChangesRocksdb) -> Result { + let BlockRocksdb { header, transactions } = block; + let total = transactions.len() + changes.account_changes.len() + changes.slot_changes.len(); + let mut paginator = self.cursor_paginator(total, header.hash.into())?; + + let tx_range = paginator.take(transactions.len()); + let account_entries = sorted_account_changes(&changes); + let account_range = paginator.take(account_entries.len()); + let slot_entries = sorted_slot_changes(&changes); + let slot_range = paginator.take(slot_entries.len()); + + let mut page_changes = BlockChangesRocksdb::with_capacity(account_range.len()); + for (address, change) in account_entries[account_range].iter().cloned() { + page_changes.account_changes.insert(address, change); + } + for ((address, slot), value) in slot_entries[slot_range].iter().copied() { + page_changes.slot_changes.insert((address, slot), value); + } + + Ok(BlockWithChangesPageResponse { + block: BlockRocksdb { + header, + transactions: transactions[tx_range].to_vec(), + }, + changes: page_changes, + pagination: paginator.finish(), + }) + } + + fn resolve_filter(filter: BlockFilter, cursor: Option<&str>) -> Result<(BlockFilter, usize), RpcError> { + match cursor { + Some(cursor) => { + let (cursor, next_index) = BlockHashCursor::decode_cursor(cursor)?; + Ok((BlockFilter::Hash(cursor.block_hash), next_index)) + } + None => Ok((filter, 0)), + } + } + + fn cursor_paginator(&self, total: usize, block_hash: Hash) -> Result { + ImporterCursorPaginator::new( + total, + self.start, + self.request.limit(), + CursorPaginationPolicy::new(BlockHashCursor { block_hash }), + ) + } +} + +pub type ImporterPageInfo = CursorPageInfo; + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BlockAndReceiptsPageResponse { + pub block: JsonValue, + pub receipts: Vec, + pub pagination: ImporterPageInfo, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BlockWithChangesPageResponse { + pub block: BlockRocksdb, + pub changes: BlockChangesRocksdb, + pub pagination: ImporterPageInfo, +} + +fn sorted_account_changes(changes: &BlockChangesRocksdb) -> Vec<(AddressRocksdb, AccountChangesRocksdb)> { + let mut entries = changes + .account_changes + .iter() + .map(|(address, change)| (*address, change.clone())) + .collect::>(); + entries.sort_by_key(|(address, _)| *address); + entries +} + +fn sorted_slot_changes(changes: &BlockChangesRocksdb) -> Vec<((AddressRocksdb, SlotIndexRocksdb), SlotValueRocksdb)> { + let mut entries = changes.slot_changes.iter().map(|(key, value)| (*key, *value)).collect::>(); + entries.sort_by_key(|(key, _)| *key); + entries +} + +pub(crate) type ImporterCursorPaginator = CursorPaginator; + +pub(crate) struct BlockHashCursor { + block_hash: Hash, +} + +impl CursorCodec for BlockHashCursor { + type Error = RpcError; + + fn invalid_start_error() -> Self::Error { + RpcError::ParameterInvalid + } + + fn encode_cursor(&self, next_index: usize) -> String { + format!("{IMPORTER_CURSOR_VERSION}:{}:{next_index}", self.block_hash) + } + + fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error> { + let mut parts = cursor.split(':'); + let version = parts.next().ok_or(RpcError::ParameterInvalid)?; + let block_hash = parts.next().ok_or(RpcError::ParameterInvalid)?; + let next_index = parts.next().ok_or(RpcError::ParameterInvalid)?; + + if version != IMPORTER_CURSOR_VERSION || parts.next().is_some() { + return Err(RpcError::ParameterInvalid); + } + + Ok(( + Self { + block_hash: block_hash.parse().map_err(|_| RpcError::ParameterInvalid)?, + }, + next_index.parse().map_err(|_| RpcError::ParameterInvalid)?, + )) + } +} diff --git a/src/eth/rpc/types/mod.rs b/src/eth/rpc/types/mod.rs index 52bbeac38..656b3d176 100644 --- a/src/eth/rpc/types/mod.rs +++ b/src/eth/rpc/types/mod.rs @@ -1,16 +1,27 @@ mod block_filter; mod error; +mod importer_pagination; mod log_filter; mod log_filter_input; +mod pagination; mod rpc_client_app; mod timestamp_filter; pub use block_filter::BlockFilter; pub use error::MulticallError; pub use error::RpcError; +pub use importer_pagination::BlockAndReceiptsPageResponse; +pub use importer_pagination::BlockWithChangesPageResponse; +pub use importer_pagination::IMPORTER_PAGE_LIMIT_DEFAULT; +pub(crate) use importer_pagination::ImporterCursorPaginator; +pub use importer_pagination::ImporterPageInfo; +pub use importer_pagination::ImporterPageRequest; +pub use importer_pagination::ImporterPagination; pub use log_filter::LogFilter; pub use log_filter_input::LogFilterInput; pub use log_filter_input::LogFilterInputTopic; +pub use pagination::PageReducer; +pub use pagination::PaginatedPageFetcher; pub use rpc_client_app::RpcClientApp; pub use timestamp_filter::BlockTimestampFilter; pub use timestamp_filter::BlockTimestampSeekMode; diff --git a/src/eth/rpc/types/pagination.rs b/src/eth/rpc/types/pagination.rs new file mode 100644 index 000000000..d99187a6d --- /dev/null +++ b/src/eth/rpc/types/pagination.rs @@ -0,0 +1,220 @@ +use std::future::Future; +use std::marker::PhantomData; +use std::ops::Range; + +// TODO: test module + +/// Immutable result produced by the pagination engine after a page is built. +pub struct PageOutcome { + pub limit: usize, + pub returned: usize, + pub total: usize, + pub next_page: Option, +} + +/// Generic interface exposed by pagination engines. +pub trait Paginator { + type Error; + type NextPage; + type PageInfo; + + fn take(&mut self, section_len: usize) -> Range; + fn finish(self) -> Self::PageInfo; +} + +/// Defines how a paginator validates requests and converts engine output into +/// a domain-specific page-info response. +pub trait PaginationPolicy { + type Error; + type NextPage; + type PageInfo; + + fn invalid_start_error() -> Self::Error; + fn next_page(&self, next_index: usize) -> Self::NextPage; + fn page_info(&self, outcome: PageOutcome) -> Self::PageInfo; +} + +/// Generic pagination engine. +/// +/// The engine only knows how to slice a logical stream. The policy decides how +/// invalid requests are reported and how the final page information is encoded +/// (cursor, limit/offset, page number, etc.). +pub struct PaginatorEngine { + policy: Policy, + start: usize, + limit: usize, + total: usize, + returned: usize, + skipped: usize, + remaining: usize, +} + +impl PaginatorEngine +where + Policy: PaginationPolicy, +{ + pub fn new(total: usize, start: usize, limit: usize, policy: Policy) -> Result { + if start > total || (start == total && total != 0) { + return Err(Policy::invalid_start_error()); + } + + Ok(Self { + policy, + start, + limit, + total, + returned: 0, + skipped: start, + remaining: limit, + }) + } + + fn next_index(&self) -> Option { + let next_index = self.start.saturating_add(self.returned); + (next_index < self.total).then_some(next_index) + } +} + +impl Paginator for PaginatorEngine +where + Policy: PaginationPolicy, +{ + type Error = Policy::Error; + type NextPage = Policy::NextPage; + type PageInfo = Policy::PageInfo; + + fn take(&mut self, section_len: usize) -> Range { + let range_start = self.skipped.min(section_len); + + let range_end = if self.skipped < section_len { + range_start.saturating_add(self.remaining).min(section_len) + } else { + range_start + }; + + let returned = range_end.saturating_sub(range_start); + + self.returned += returned; + self.remaining = self.remaining.saturating_sub(returned); + self.skipped = self.skipped.saturating_sub(section_len); + + range_start..range_end + } + + fn finish(self) -> Self::PageInfo { + let next_page = self.next_index().map(|next_index| self.policy.next_page(next_index)); + self.policy.page_info(PageOutcome { + limit: self.limit, + returned: self.returned, + total: self.total, + next_page, + }) + } +} + +/// Converts a domain cursor value into the generic cursor-pagination policy. +pub trait CursorCodec: Sized { + type Error; + + fn invalid_start_error() -> Self::Error; + fn encode_cursor(&self, next_index: usize) -> String; + fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error>; +} + +/// Reusable cursor pagination policy. Domains provide only the cursor payload +/// codec; the paginator behavior remains shared. +pub struct CursorPaginationPolicy { + cursor: Cursor, +} + +impl CursorPaginationPolicy { + pub fn new(cursor: Cursor) -> Self { + Self { cursor } + } +} + +impl PaginationPolicy for CursorPaginationPolicy +where + Cursor: CursorCodec, +{ + type Error = Cursor::Error; + type NextPage = String; + type PageInfo = CursorPageInfo; + + fn invalid_start_error() -> Self::Error { + Cursor::invalid_start_error() + } + + fn next_page(&self, next_index: usize) -> Self::NextPage { + self.cursor.encode_cursor(next_index) + } + + fn page_info(&self, outcome: PageOutcome) -> Self::PageInfo { + CursorPageInfo { + limit: outcome.limit, + returned: outcome.returned, + total: outcome.total, + next_cursor: outcome.next_page, + } + } +} + +/// Generic cursor paginator for every domain-specific cursor payload. +pub type CursorPaginator = PaginatorEngine>; + +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CursorPageInfo { + pub limit: usize, + pub returned: usize, + pub total: usize, + pub next_cursor: Option, +} + +/// Reducer for client-side paginated fetches. +/// +/// Every page is folded into an accumulator until the paginator reports +/// no next page, then the accumulator is finalized. +pub trait PageReducer { + type Output; + type Paginator: Paginator; + + fn reduce(&mut self, page: Page) -> anyhow::Result::NextPage>>; + fn finish_after_not_found(self) -> anyhow::Result>; + fn finish(self) -> anyhow::Result>; +} + +/// Client-side fetcher that repeatedly fetches pages and reduces them into a final output. +pub struct PaginatedPageFetcher { + reducer: Reducer, + _paginator: PhantomData, +} + +impl PaginatedPageFetcher { + pub fn new(reducer: Reducer) -> Self { + Self { + reducer, + _paginator: PhantomData, + } + } + + pub async fn collect(mut self, mut fetch_page: FetchPage) -> anyhow::Result> + where + Reducer: PageReducer, + FetchPage: FnMut(Option<::NextPage>) -> FetchFuture, + FetchFuture: Future>>, + { + let mut next_page = None; + + loop { + let Some(page) = fetch_page(next_page).await? else { + return self.reducer.finish_after_not_found(); + }; + + next_page = self.reducer.reduce(page)?; + if next_page.is_none() { + return self.reducer.finish(); + } + } + } +} diff --git a/src/eth/types/external/external_block.rs b/src/eth/types/external/external_block.rs index eb4be6198..71f844723 100644 --- a/src/eth/types/external/external_block.rs +++ b/src/eth/types/external/external_block.rs @@ -10,6 +10,8 @@ use alloy_primitives::Bloom; use alloy_primitives::Bytes; #[cfg(test)] use alloy_primitives::U256; +use alloy_rpc_types_eth::BlockTransactions; +use anyhow::bail; #[cfg(test)] use fake::Dummy; #[cfg(test)] @@ -55,6 +57,36 @@ impl ExternalBlock { pub fn author(&self) -> Address { self.0.header.inner.beneficiary.into() } + + /// Returns the number of full transactions in the block. + pub fn full_transactions_len(&self) -> anyhow::Result { + let BlockTransactions::Full(transactions) = &self.0.transactions else { + bail!("expected full transactions, got hashes or uncle"); + }; + + Ok(transactions.len()) + } + + /// Appends full transactions from another page of the same block. + pub fn extend_full_transactions_from(&mut self, other: Self) -> anyhow::Result<()> { + if self.hash() != other.hash() { + bail!( + "cannot extend external block transactions from block {} into block {}", + other.hash(), + self.hash() + ); + } + + let BlockTransactions::Full(other_transactions) = other.0.transactions else { + bail!("expected full transactions, got hashes or uncle"); + }; + let BlockTransactions::Full(transactions) = &mut self.0.transactions else { + bail!("expected full transactions, got hashes or uncle"); + }; + + transactions.extend(other_transactions); + Ok(()) + } } impl PartialEq for ExternalBlock { From 7c976e51078497922ce08c9336a89f32b16a1b79 Mon Sep 17 00:00:00 2001 From: gventino-cw Date: Thu, 20 Aug 2026 18:10:32 -0300 Subject: [PATCH 2/3] refac: simplifying paginator interfaces --- .../blockchain_client/importer_pagination.rs | 5 +- src/eth/rpc/server.rs | 4 +- src/eth/rpc/types/importer_pagination.rs | 79 +++++- src/eth/rpc/types/mod.rs | 1 - src/eth/rpc/types/pagination.rs | 267 +++++++++++------- 5 files changed, 236 insertions(+), 120 deletions(-) diff --git a/src/eth/rpc/blockchain_client/importer_pagination.rs b/src/eth/rpc/blockchain_client/importer_pagination.rs index 70fff7029..32ee5f95c 100644 --- a/src/eth/rpc/blockchain_client/importer_pagination.rs +++ b/src/eth/rpc/blockchain_client/importer_pagination.rs @@ -6,7 +6,6 @@ use super::blockchain_client::BlockchainClient; use crate::eth::rpc::types::BlockAndReceiptsPageResponse; use crate::eth::rpc::types::BlockWithChangesPageResponse; use crate::eth::rpc::types::IMPORTER_PAGE_LIMIT_DEFAULT; -use crate::eth::rpc::types::ImporterCursorPaginator; use crate::eth::rpc::types::ImporterPageInfo; use crate::eth::rpc::types::ImporterPageRequest; use crate::eth::rpc::types::PageReducer; @@ -117,7 +116,7 @@ impl BlockAndReceiptsPages { impl PageReducer for BlockAndReceiptsPages { type Output = ExternalBlockWithReceipts; - type Paginator = ImporterCursorPaginator; + type NextPage = String; fn reduce(&mut self, page: BlockAndReceiptsPageResponse) -> anyhow::Result> { let cursor = validate_progress(&page.pagination, &mut self.expected_total, "block with receipts")?; @@ -216,7 +215,7 @@ impl BlockWithChangesPages { impl PageReducer for BlockWithChangesPages { type Output = (BlockRocksdb, BlockChangesRocksdb); - type Paginator = ImporterCursorPaginator; + type NextPage = String; fn reduce(&mut self, page: BlockWithChangesPageResponse) -> anyhow::Result> { let cursor = validate_progress(&page.pagination, &mut self.expected_total, "block with changes")?; diff --git a/src/eth/rpc/server.rs b/src/eth/rpc/server.rs index bfd06a879..58e049f34 100644 --- a/src/eth/rpc/server.rs +++ b/src/eth/rpc/server.rs @@ -899,7 +899,7 @@ fn stratus_get_block_and_receipts(params: Params<'_>, ctx: Arc, ext: // parse params let (params, filter) = next_rpc_param::(params.sequence())?; - let pagination = ImporterPagination::next(params, filter)?; + let pagination = ImporterPagination::from_params(params, filter)?; // track tracing::info!(%filter, "reading block and receipts"); @@ -936,7 +936,7 @@ fn stratus_get_block_with_changes(params: Params<'_>, ctx: Arc, ext: // parse params let (params, filter) = next_rpc_param::(params.sequence())?; - let pagination = ImporterPagination::next(params, filter)?; + let pagination = ImporterPagination::from_params(params, filter)?; // track tracing::info!(%filter, "reading block and changes"); diff --git a/src/eth/rpc/types/importer_pagination.rs b/src/eth/rpc/types/importer_pagination.rs index 77bb69a6e..b6f17a735 100644 --- a/src/eth/rpc/types/importer_pagination.rs +++ b/src/eth/rpc/types/importer_pagination.rs @@ -4,7 +4,6 @@ use super::BlockFilter; use super::RpcError; use super::pagination::CursorCodec; use super::pagination::CursorPageInfo; -use super::pagination::CursorPaginationPolicy; use super::pagination::CursorPaginator; use super::pagination::Paginator; use crate::alias::AlloyReceipt; @@ -56,7 +55,7 @@ pub struct ImporterPagination { } impl ImporterPagination { - pub fn next(params: ParamsSequence<'_>, filter: BlockFilter) -> Result, RpcError> { + pub fn from_params(params: ParamsSequence<'_>, filter: BlockFilter) -> Result, RpcError> { let Some(request) = ImporterPageRequest::parse_next(params)? else { return Ok(None); }; @@ -120,12 +119,7 @@ impl ImporterPagination { } fn cursor_paginator(&self, total: usize, block_hash: Hash) -> Result { - ImporterCursorPaginator::new( - total, - self.start, - self.request.limit(), - CursorPaginationPolicy::new(BlockHashCursor { block_hash }), - ) + ImporterCursorPaginator::new(total, self.start, self.request.limit(), BlockHashCursor { block_hash }) } } @@ -163,7 +157,7 @@ fn sorted_slot_changes(changes: &BlockChangesRocksdb) -> Vec<((AddressRocksdb, S entries } -pub(crate) type ImporterCursorPaginator = CursorPaginator; +type ImporterCursorPaginator = CursorPaginator; pub(crate) struct BlockHashCursor { block_hash: Hash, @@ -198,3 +192,70 @@ impl CursorCodec for BlockHashCursor { )) } } + +#[cfg(test)] +mod tests { + use super::BlockHashCursor; + use super::CursorCodec; + use super::RpcError; + use crate::eth::types::Hash; + + fn codec(block_hash: &str) -> BlockHashCursor { + BlockHashCursor { block_hash: block_hash.parse().unwrap() } + } + + #[test] + fn cursor_roundtrip_preserves_hash_and_index() { + let original = codec("0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c"); + let encoded = original.encode_cursor(42); + let (decoded, next_index) = BlockHashCursor::decode_cursor(&encoded).expect("decode succeeds"); + + assert_eq!(decoded.block_hash, original.block_hash); + assert_eq!(next_index, 42); + } + + #[test] + fn cursor_decode_rejects_wrong_version() { + let bad = "v2:0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c:0"; + let err = BlockHashCursor::decode_cursor(bad).err(); + assert!(matches!(err, Some(RpcError::ParameterInvalid))); + } + + #[test] + fn cursor_decode_rejects_missing_parts() { + assert!(BlockHashCursor::decode_cursor("v1:0xabc").is_err()); + assert!(BlockHashCursor::decode_cursor("v1").is_err()); + } + + #[test] + fn cursor_decode_rejects_extra_parts() { + let extra = "v1:0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c:0:extra"; + assert!(BlockHashCursor::decode_cursor(extra).is_err()); + } + + #[test] + fn cursor_decode_rejects_non_numeric_index() { + let bad = "v1:0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c:notanumber"; + assert!(BlockHashCursor::decode_cursor(bad).is_err()); + } + + #[test] + fn cursor_decode_rejects_invalid_hash() { + let bad = "v1:0xnotahash:0"; + assert!(BlockHashCursor::decode_cursor(bad).is_err()); + } + + #[test] + fn encode_uses_v1_format() { + let c = codec("0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c"); + let encoded = c.encode_cursor(7); + assert!(encoded.starts_with("v1:")); + assert_eq!(encoded.split(':').count(), 3); + } + + // ensure Hash parses from a 0x-prefixed hex string in tests + #[test] + fn hash_parses_for_test_fixture() { + let _: Hash = "0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c".parse().unwrap(); + } +} diff --git a/src/eth/rpc/types/mod.rs b/src/eth/rpc/types/mod.rs index 656b3d176..84a730e7f 100644 --- a/src/eth/rpc/types/mod.rs +++ b/src/eth/rpc/types/mod.rs @@ -13,7 +13,6 @@ pub use error::RpcError; pub use importer_pagination::BlockAndReceiptsPageResponse; pub use importer_pagination::BlockWithChangesPageResponse; pub use importer_pagination::IMPORTER_PAGE_LIMIT_DEFAULT; -pub(crate) use importer_pagination::ImporterCursorPaginator; pub use importer_pagination::ImporterPageInfo; pub use importer_pagination::ImporterPageRequest; pub use importer_pagination::ImporterPagination; diff --git a/src/eth/rpc/types/pagination.rs b/src/eth/rpc/types/pagination.rs index d99187a6d..e5aa71c29 100644 --- a/src/eth/rpc/types/pagination.rs +++ b/src/eth/rpc/types/pagination.rs @@ -1,18 +1,10 @@ use std::future::Future; -use std::marker::PhantomData; use std::ops::Range; -// TODO: test module - -/// Immutable result produced by the pagination engine after a page is built. -pub struct PageOutcome { - pub limit: usize, - pub returned: usize, - pub total: usize, - pub next_page: Option, -} - /// Generic interface exposed by pagination engines. +/// +/// Kept as a trait so alternative slicing strategies (limit/offset, page number, etc.) +/// can be slotted in later without touching call sites. pub trait Paginator { type Error; type NextPage; @@ -22,25 +14,24 @@ pub trait Paginator { fn finish(self) -> Self::PageInfo; } -/// Defines how a paginator validates requests and converts engine output into -/// a domain-specific page-info response. -pub trait PaginationPolicy { +/// Converts a domain cursor value into the generic cursor-pagination behavior. +/// +/// Domains implement only this codec; the paginator behavior is shared. +pub trait CursorCodec: Sized { type Error; - type NextPage; - type PageInfo; fn invalid_start_error() -> Self::Error; - fn next_page(&self, next_index: usize) -> Self::NextPage; - fn page_info(&self, outcome: PageOutcome) -> Self::PageInfo; + fn encode_cursor(&self, next_index: usize) -> String; + fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error>; } -/// Generic pagination engine. +/// Cursor-based paginator. /// -/// The engine only knows how to slice a logical stream. The policy decides how -/// invalid requests are reported and how the final page information is encoded -/// (cursor, limit/offset, page number, etc.). -pub struct PaginatorEngine { - policy: Policy, +/// Knows only how to slice a logical stream. The [`CursorCodec`] decides how +/// cursors are encoded/decoded; the paginator reports the final page info via +/// [`CursorPageInfo`]. +pub struct CursorPaginator { + codec: C, start: usize, limit: usize, total: usize, @@ -49,17 +40,17 @@ pub struct PaginatorEngine { remaining: usize, } -impl PaginatorEngine +impl CursorPaginator where - Policy: PaginationPolicy, + C: CursorCodec, { - pub fn new(total: usize, start: usize, limit: usize, policy: Policy) -> Result { + pub fn new(total: usize, start: usize, limit: usize, codec: C) -> Result { if start > total || (start == total && total != 0) { - return Err(Policy::invalid_start_error()); + return Err(C::invalid_start_error()); } Ok(Self { - policy, + codec, start, limit, total, @@ -75,93 +66,44 @@ where } } -impl Paginator for PaginatorEngine +impl Paginator for CursorPaginator where - Policy: PaginationPolicy, + C: CursorCodec, { - type Error = Policy::Error; - type NextPage = Policy::NextPage; - type PageInfo = Policy::PageInfo; + type Error = C::Error; + type NextPage = String; + type PageInfo = CursorPageInfo; fn take(&mut self, section_len: usize) -> Range { - let range_start = self.skipped.min(section_len); + // skip an entire section + if self.skipped >= section_len { + self.skipped -= section_len; + return section_len..section_len; + } - let range_end = if self.skipped < section_len { - range_start.saturating_add(self.remaining).min(section_len) - } else { - range_start - }; + let start = self.skipped; + self.skipped = 0; - let returned = range_end.saturating_sub(range_start); + let end = (start + self.remaining).min(section_len); + let returned = end - start; self.returned += returned; - self.remaining = self.remaining.saturating_sub(returned); - self.skipped = self.skipped.saturating_sub(section_len); + self.remaining -= returned; - range_start..range_end + start..end } fn finish(self) -> Self::PageInfo { - let next_page = self.next_index().map(|next_index| self.policy.next_page(next_index)); - self.policy.page_info(PageOutcome { + let next_cursor = self.next_index().map(|next_index| self.codec.encode_cursor(next_index)); + CursorPageInfo { limit: self.limit, returned: self.returned, total: self.total, - next_page, - }) - } -} - -/// Converts a domain cursor value into the generic cursor-pagination policy. -pub trait CursorCodec: Sized { - type Error; - - fn invalid_start_error() -> Self::Error; - fn encode_cursor(&self, next_index: usize) -> String; - fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error>; -} - -/// Reusable cursor pagination policy. Domains provide only the cursor payload -/// codec; the paginator behavior remains shared. -pub struct CursorPaginationPolicy { - cursor: Cursor, -} - -impl CursorPaginationPolicy { - pub fn new(cursor: Cursor) -> Self { - Self { cursor } - } -} - -impl PaginationPolicy for CursorPaginationPolicy -where - Cursor: CursorCodec, -{ - type Error = Cursor::Error; - type NextPage = String; - type PageInfo = CursorPageInfo; - - fn invalid_start_error() -> Self::Error { - Cursor::invalid_start_error() - } - - fn next_page(&self, next_index: usize) -> Self::NextPage { - self.cursor.encode_cursor(next_index) - } - - fn page_info(&self, outcome: PageOutcome) -> Self::PageInfo { - CursorPageInfo { - limit: outcome.limit, - returned: outcome.returned, - total: outcome.total, - next_cursor: outcome.next_page, + next_cursor, } } } -/// Generic cursor paginator for every domain-specific cursor payload. -pub type CursorPaginator = PaginatorEngine>; - #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] pub struct CursorPageInfo { @@ -177,9 +119,9 @@ pub struct CursorPageInfo { /// no next page, then the accumulator is finalized. pub trait PageReducer { type Output; - type Paginator: Paginator; + type NextPage; - fn reduce(&mut self, page: Page) -> anyhow::Result::NextPage>>; + fn reduce(&mut self, page: Page) -> anyhow::Result>; fn finish_after_not_found(self) -> anyhow::Result>; fn finish(self) -> anyhow::Result>; } @@ -187,21 +129,17 @@ pub trait PageReducer { /// Client-side fetcher that repeatedly fetches pages and reduces them into a final output. pub struct PaginatedPageFetcher { reducer: Reducer, - _paginator: PhantomData, } impl PaginatedPageFetcher { pub fn new(reducer: Reducer) -> Self { - Self { - reducer, - _paginator: PhantomData, - } + Self { reducer } } pub async fn collect(mut self, mut fetch_page: FetchPage) -> anyhow::Result> where Reducer: PageReducer, - FetchPage: FnMut(Option<::NextPage>) -> FetchFuture, + FetchPage: FnMut(Option) -> FetchFuture, FetchFuture: Future>>, { let mut next_page = None; @@ -218,3 +156,122 @@ impl PaginatedPageFetcher { } } } + +#[cfg(test)] +mod tests { + use std::ops::Range; + + use super::CursorCodec; + use super::CursorPageInfo; + use super::CursorPaginator; + use super::Paginator; + + // A minimal codec for testing the paginator independently of any domain type. + struct IndexCodec; + + impl CursorCodec for IndexCodec { + type Error = &'static str; + + fn invalid_start_error() -> Self::Error { + "invalid start" + } + + fn encode_cursor(&self, next_index: usize) -> String { + next_index.to_string() + } + + fn decode_cursor(cursor: &str) -> Result<(Self, usize), Self::Error> { + Ok((IndexCodec, cursor.parse().map_err(|_| "bad cursor")?)) + } + } + + fn paginator(total: usize, start: usize, limit: usize) -> CursorPaginator { + CursorPaginator::new(total, start, limit, IndexCodec).expect("valid paginator") + } + + fn take_all(mut p: CursorPaginator, sections: &[usize]) -> (Vec>, CursorPageInfo) { + let mut ranges = Vec::new(); + for §ion_len in sections { + ranges.push(p.take(section_len)); + } + let info = p.finish(); + (ranges, info) + } + + #[test] + fn empty_total_is_valid() { + let p = paginator(0, 0, 10); + let (_, info) = take_all(p, &[]); + assert_eq!(info, CursorPageInfo { limit: 10, returned: 0, total: 0, next_cursor: None }); + } + + #[test] + fn start_equal_to_total_is_invalid() { + assert!(CursorPaginator::new(5, 5, 10, IndexCodec).is_err()); + } + + #[test] + fn start_greater_than_total_is_invalid() { + assert!(CursorPaginator::new(5, 6, 10, IndexCodec).is_err()); + } + + #[test] + fn limit_smaller_than_one_section() { + let (ranges, info) = take_all(paginator(100, 0, 3), &[10]); + assert_eq!(ranges, vec![0..3]); + assert_eq!(info.returned, 3); + assert_eq!(info.next_cursor, Some("3".to_string())); + } + + #[test] + fn limit_spanning_multiple_sections() { + let (ranges, info) = take_all(paginator(100, 0, 7), &[3, 3, 3]); + // first section fully taken (3), second fully taken (3), third partial (1) + assert_eq!(ranges, vec![0..3, 0..3, 0..1]); + assert_eq!(info.returned, 7); + assert_eq!(info.next_cursor, Some("7".to_string())); + } + + #[test] + fn skip_across_section_boundary() { + // start=5, sections of length 3, 3, 3 -> skip 3 (section 1), skip 2 (section 2), + // then take 1 from section 2 and the remaining 3 from section 3. + let (ranges, info) = take_all(paginator(100, 5, 4), &[3, 3, 3]); + assert_eq!(ranges, vec![3..3, 2..3, 0..3]); + assert_eq!(info.returned, 4); + assert_eq!(info.next_cursor, Some("9".to_string())); + } + + #[test] + fn fully_consumed_has_no_cursor() { + let (ranges, info) = take_all(paginator(5, 0, 5), &[3, 2]); + assert_eq!(ranges, vec![0..3, 0..2]); + assert_eq!(info.returned, 5); + assert_eq!(info.next_cursor, None); + } + + #[test] + fn limit_larger_than_total_clamps_to_total() { + let (ranges, info) = take_all(paginator(4, 0, 100), &[2, 2]); + assert_eq!(ranges, vec![0..2, 0..2]); + assert_eq!(info.returned, 4); + assert_eq!(info.next_cursor, None); + assert_eq!(info.total, 4); + } + + #[test] + fn section_larger_than_remaining_returns_partial() { + let (ranges, info) = take_all(paginator(100, 0, 2), &[10]); + assert_eq!(ranges, vec![0..2]); + assert_eq!(info.returned, 2); + assert_eq!(info.next_cursor, Some("2".to_string())); + } + + #[test] + fn extra_sections_after_limit_are_empty() { + let (ranges, info) = take_all(paginator(100, 0, 2), &[2, 2, 2]); + assert_eq!(ranges, vec![0..2, 0..0, 0..0]); + assert_eq!(info.returned, 2); + assert_eq!(info.next_cursor, Some("2".to_string())); + } +} From f17e2dbb3051c9f41af036f33aa25e0b0f27f803 Mon Sep 17 00:00:00 2001 From: gventino-cw Date: Thu, 20 Aug 2026 18:49:34 -0300 Subject: [PATCH 3/3] feat: tests --- .../blockchain_client/importer_pagination.rs | 367 ++++++++++++++++++ src/eth/rpc/types/importer_pagination.rs | 231 ++++++++++- src/eth/rpc/types/pagination.rs | 131 ++++++- src/eth/types/external/external_block.rs | 104 +++++ 4 files changed, 830 insertions(+), 3 deletions(-) diff --git a/src/eth/rpc/blockchain_client/importer_pagination.rs b/src/eth/rpc/blockchain_client/importer_pagination.rs index 32ee5f95c..1dc737a1f 100644 --- a/src/eth/rpc/blockchain_client/importer_pagination.rs +++ b/src/eth/rpc/blockchain_client/importer_pagination.rs @@ -246,3 +246,370 @@ impl PageReducer for BlockWithChangesPages { Ok(Some((block, self.changes))) } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use fake::Fake; + use fake::Faker; + use hash_hasher::HashBuildHasher; + + use super::BlockAndReceiptsPages; + use super::BlockWithChangesPages; + use super::validate_progress; + use crate::eth::rpc::types::BlockAndReceiptsPageResponse; + use crate::eth::rpc::types::BlockWithChangesPageResponse; + use crate::eth::rpc::types::ImporterPageInfo; + use crate::eth::rpc::types::PageReducer; + use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb; + use crate::eth::storage::permanent::rocks::types::AddressRocksdb; + use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; + use crate::eth::storage::permanent::rocks::types::BlockRocksdb; + use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb; + use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb; + use crate::eth::types::Block; + use crate::eth::types::BlockNumber; + use crate::eth::types::ExternalBlock; + use crate::eth::types::ExternalReceipt; + use crate::eth::types::SlotIndex; + use crate::eth::types::SlotValue; + use crate::eth::types::UnixTime; + use crate::ext::to_json_value; + + // helpers + + fn page_info(returned: usize, total: usize, next_cursor: Option<&str>) -> ImporterPageInfo { + ImporterPageInfo { + limit: 256, + returned, + total, + next_cursor: next_cursor.map(String::from), + } + } + + fn external_block_with_txs(count: usize) -> ExternalBlock { + let mut block: ExternalBlock = Faker.fake(); + let txs: Vec<_> = std::iter::repeat_with(|| Faker.fake()).take(count).collect(); + block.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(txs); + block + } + + fn split_external_block(block: &ExternalBlock, at: usize) -> (ExternalBlock, ExternalBlock) { + let txs = match &block.0.transactions { + alloy_rpc_types_eth::BlockTransactions::Full(txs) => txs.clone(), + _ => unreachable!(), + }; + let mut page1 = block.clone(); + let mut page2 = block.clone(); + page1.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(txs[..at].to_vec()); + page2.0.transactions = alloy_rpc_types_eth::BlockTransactions::Full(txs[at..].to_vec()); + (page1, page2) + } + + fn receipts_page(block: ExternalBlock, receipts: Vec, pagination: ImporterPageInfo) -> BlockAndReceiptsPageResponse { + BlockAndReceiptsPageResponse { + block: to_json_value(block), + receipts, + pagination, + } + } + + fn block_rocksdb_with_txs(number: u64, count: usize) -> BlockRocksdb { + let mut block = Block::new(BlockNumber::from(number), UnixTime::from(0u64)); + block.transactions = std::iter::repeat_with(|| Faker.fake()).take(count).collect(); + BlockRocksdb::from(block) + } + + fn changes_with_accounts(addresses: &[[u8; 20]]) -> BlockChangesRocksdb { + let mut account_changes = HashMap::with_hasher(HashBuildHasher::default()); + for addr in addresses { + account_changes.insert(AddressRocksdb(*addr), AccountChangesRocksdb::default()); + } + BlockChangesRocksdb { + account_changes, + slot_changes: HashMap::with_hasher(HashBuildHasher::default()), + } + } + + fn changes_with_slots(slots: &[(AddressRocksdb, SlotIndexRocksdb)]) -> BlockChangesRocksdb { + let mut slot_changes = HashMap::with_hasher(HashBuildHasher::default()); + for (addr, idx) in slots { + slot_changes.insert((*addr, *idx), SlotValueRocksdb::default()); + } + BlockChangesRocksdb { + account_changes: HashMap::with_hasher(HashBuildHasher::default()), + slot_changes, + } + } + + fn slot_idx(a: u64, b: u64, c: u64, d: u64) -> SlotIndexRocksdb { + SlotIndexRocksdb::from(SlotIndex::from([a, b, c, d])) + } + + fn slot_val(a: u64, b: u64, c: u64, d: u64) -> SlotValueRocksdb { + SlotValueRocksdb::from(SlotValue::from([a, b, c, d])) + } + + // validate_progress + + #[test] + fn validate_progress_first_page_sets_expected_total() { + let mut expected_total = None; + let info = page_info(3, 10, Some("cursor")); + let cursor = validate_progress(&info, &mut expected_total, "test").expect("ok"); + assert_eq!(expected_total, Some(10)); + assert_eq!(cursor, Some("cursor".to_string())); + } + + #[test] + fn validate_progress_same_total_ok() { + let mut expected_total = Some(10); + let info = page_info(3, 10, Some("cursor")); + let cursor = validate_progress(&info, &mut expected_total, "test").expect("ok"); + assert_eq!(cursor, Some("cursor".to_string())); + } + + #[test] + fn validate_progress_different_total_errors() { + let mut expected_total = Some(10); + let info = page_info(3, 20, Some("cursor")); + assert!(validate_progress(&info, &mut expected_total, "test").is_err()); + } + + #[test] + fn validate_progress_zero_returned_with_cursor_errors() { + let mut expected_total = None; + let info = page_info(0, 10, Some("cursor")); + assert!(validate_progress(&info, &mut expected_total, "test").is_err()); + } + + // BlockAndReceiptsPages reducer + + #[test] + fn receipts_reducer_two_pages_reassemble() { + let full_block = external_block_with_txs(5); + let (page1_block, page2_block) = split_external_block(&full_block, 3); + let block_number = full_block.number(); + + let mut reducer = BlockAndReceiptsPages::new(block_number); + + let page1 = receipts_page(page1_block, vec![Faker.fake(); 3], page_info(3, 5, Some("cursor"))); + let cursor = reducer.reduce(page1).expect("ok"); + assert_eq!(cursor, Some("cursor".to_string())); + + let page2 = receipts_page(page2_block, vec![Faker.fake(); 2], page_info(2, 5, None)); + let cursor = reducer.reduce(page2).expect("ok"); + assert!(cursor.is_none()); + + let result = reducer.finish().expect("ok").expect("some output"); + assert_eq!(result.block.full_transactions_len().unwrap(), 5); + assert_eq!(result.receipts.len(), 5); + } + + #[test] + fn receipts_reducer_wrong_block_number_errors() { + let block = external_block_with_txs(2); + let block_number = BlockNumber::from(999u64); + + let mut reducer = BlockAndReceiptsPages::new(block_number); + let page = receipts_page(block, vec![], page_info(0, 0, None)); + assert!(reducer.reduce(page).is_err()); + } + + #[test] + fn receipts_reducer_count_mismatch_errors() { + let block = external_block_with_txs(3); + let block_number = block.number(); + + let mut reducer = BlockAndReceiptsPages::new(block_number); + let page = receipts_page(block, vec![Faker.fake(); 2], page_info(3, 3, None)); + reducer.reduce(page).expect("ok"); + + assert!(reducer.finish().is_err()); + } + + #[test] + fn receipts_reducer_finish_after_not_found_no_block_returns_none() { + let reducer = BlockAndReceiptsPages::new(BlockNumber::from(1u64)); + assert!(reducer.finish_after_not_found().expect("ok").is_none()); + } + + #[test] + fn receipts_reducer_finish_after_not_found_with_partial_block_errors() { + let block = external_block_with_txs(1); + let mut reducer = BlockAndReceiptsPages::new(block.number()); + let page = receipts_page(block, vec![Faker.fake()], page_info(1, 1, Some("cursor"))); + reducer.reduce(page).expect("ok"); + + assert!(reducer.finish_after_not_found().is_err()); + } + + // BlockWithChangesPages reducer + + #[test] + fn changes_reducer_two_pages_reassemble() { + let block_number = BlockNumber::from(1u64); + let full_block = block_rocksdb_with_txs(1, 3); + let (page1_block, page2_block) = { + let txs = full_block.transactions.clone(); + let mut p1 = full_block.clone(); + let mut p2 = full_block.clone(); + p1.transactions = txs[..1].to_vec(); + p2.transactions = txs[1..].to_vec(); + (p1, p2) + }; + + let addr_a = AddressRocksdb([0x01; 20]); + let addr_b = AddressRocksdb([0x02; 20]); + let changes1 = changes_with_accounts(&[addr_a.0]); + let changes2 = changes_with_accounts(&[addr_b.0]); + + let mut reducer = BlockWithChangesPages::new(block_number); + + let page1 = BlockWithChangesPageResponse { + block: page1_block, + changes: changes1, + pagination: page_info(2, 5, Some("cursor")), + }; + let cursor = reducer.reduce(page1).expect("ok"); + assert_eq!(cursor, Some("cursor".to_string())); + + let page2 = BlockWithChangesPageResponse { + block: page2_block, + changes: changes2, + pagination: page_info(3, 5, None), + }; + let cursor = reducer.reduce(page2).expect("ok"); + assert!(cursor.is_none()); + + let (block, changes) = reducer.finish().expect("ok").expect("some output"); + assert_eq!(block.transactions.len(), 3); + assert_eq!(changes.account_changes.len(), 2); + } + + #[test] + fn changes_reducer_block_number_mismatch_errors() { + let block = block_rocksdb_with_txs(1, 1); + let mut reducer = BlockWithChangesPages::new(BlockNumber::from(999u64)); + let page = BlockWithChangesPageResponse { + block, + changes: BlockChangesRocksdb::default(), + pagination: page_info(1, 1, None), + }; + assert!(reducer.reduce(page).is_err()); + } + + #[test] + fn changes_reducer_hash_changed_between_pages_errors() { + let block1 = block_rocksdb_with_txs(1, 1); + let mut block2 = block_rocksdb_with_txs(1, 1); + block2.header.hash = Faker.fake(); + + let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64)); + + let page1 = BlockWithChangesPageResponse { + block: block1, + changes: BlockChangesRocksdb::default(), + pagination: page_info(1, 2, Some("cursor")), + }; + reducer.reduce(page1).expect("ok"); + + let page2 = BlockWithChangesPageResponse { + block: block2, + changes: BlockChangesRocksdb::default(), + pagination: page_info(1, 2, None), + }; + assert!(reducer.reduce(page2).is_err()); + } + + #[test] + fn changes_reducer_duplicate_account_change_errors() { + let block = block_rocksdb_with_txs(1, 0); + let addr = [0x01; 20]; + let changes = changes_with_accounts(&[addr]); + + let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64)); + + let page1 = BlockWithChangesPageResponse { + block: block.clone(), + changes: changes.clone(), + pagination: page_info(1, 2, Some("cursor")), + }; + reducer.reduce(page1).expect("ok"); + + let page2 = BlockWithChangesPageResponse { + block, + changes, + pagination: page_info(1, 2, None), + }; + assert!(reducer.reduce(page2).is_err()); + } + + #[test] + fn changes_reducer_duplicate_slot_change_errors() { + let block = block_rocksdb_with_txs(1, 0); + let addr = AddressRocksdb([0x01; 20]); + let idx = slot_idx(0, 0, 0, 1); + let changes = changes_with_slots(&[(addr, idx)]); + + let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64)); + + let page1 = BlockWithChangesPageResponse { + block: block.clone(), + changes: changes.clone(), + pagination: page_info(1, 2, Some("cursor")), + }; + reducer.reduce(page1).expect("ok"); + + let page2 = BlockWithChangesPageResponse { + block, + changes, + pagination: page_info(1, 2, None), + }; + assert!(reducer.reduce(page2).is_err()); + } + + #[test] + fn changes_reducer_total_mismatch_errors() { + let block = block_rocksdb_with_txs(1, 2); + let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64)); + + let page = BlockWithChangesPageResponse { + block, + changes: BlockChangesRocksdb::default(), + pagination: page_info(2, 10, None), + }; + reducer.reduce(page).expect("ok"); + + assert!(reducer.finish().is_err()); + } + + #[test] + fn changes_reducer_finish_after_not_found_no_block_returns_none() { + let reducer = BlockWithChangesPages::new(BlockNumber::from(1u64)); + assert!(reducer.finish_after_not_found().expect("ok").is_none()); + } + + #[test] + fn changes_reducer_finish_after_not_found_with_partial_block_errors() { + let block = block_rocksdb_with_txs(1, 1); + let mut reducer = BlockWithChangesPages::new(BlockNumber::from(1u64)); + let page = BlockWithChangesPageResponse { + block, + changes: BlockChangesRocksdb::default(), + pagination: page_info(1, 1, Some("cursor")), + }; + reducer.reduce(page).expect("ok"); + + assert!(reducer.finish_after_not_found().is_err()); + } + + // Slot helpers for completeness + + #[test] + fn slot_helpers_construct_valid_values() { + let _ = slot_val(1, 2, 3, 4); + let _ = slot_idx(1, 2, 3, 4); + } +} diff --git a/src/eth/rpc/types/importer_pagination.rs b/src/eth/rpc/types/importer_pagination.rs index b6f17a735..325916169 100644 --- a/src/eth/rpc/types/importer_pagination.rs +++ b/src/eth/rpc/types/importer_pagination.rs @@ -41,7 +41,7 @@ impl ImporterPageRequest { } } - fn limit(&self) -> usize { + pub(crate) fn limit(&self) -> usize { match self.limit { Some(0) | None => IMPORTER_PAGE_LIMIT_DEFAULT, Some(limit) => limit.min(IMPORTER_PAGE_LIMIT_MAX), @@ -63,6 +63,18 @@ impl ImporterPagination { Ok(Some((filter, Self { request, start }))) } + /// Test-only constructor that builds a pagination with the given start index and limit. + #[cfg(test)] + pub(crate) fn for_test(start: usize, limit: usize) -> Self { + Self { + request: ImporterPageRequest { + cursor: None, + limit: Some(limit), + }, + start, + } + } + pub fn block_and_receipts_response(&self, block: Block) -> Result { let mut block = block; let mut paginator = self.cursor_paginator(block.transactions.len(), block.hash())?; @@ -195,15 +207,44 @@ impl CursorCodec for BlockHashCursor { #[cfg(test)] mod tests { + use std::collections::HashMap; + + use fake::Fake; + use fake::Faker; + use hash_hasher::HashBuildHasher; + use jsonrpsee::types::Params; + use super::BlockHashCursor; use super::CursorCodec; + use super::IMPORTER_PAGE_LIMIT_DEFAULT; + use super::IMPORTER_PAGE_LIMIT_MAX; + use super::ImporterPageRequest; + use super::ImporterPagination; use super::RpcError; + use crate::eth::rpc::BlockFilter; + use crate::eth::storage::permanent::rocks::types::AccountChangesRocksdb; + use crate::eth::storage::permanent::rocks::types::AddressRocksdb; + use crate::eth::storage::permanent::rocks::types::BlockChangesRocksdb; + use crate::eth::storage::permanent::rocks::types::BlockRocksdb; + use crate::eth::storage::permanent::rocks::types::SlotIndexRocksdb; + use crate::eth::storage::permanent::rocks::types::SlotValueRocksdb; + use crate::eth::types::Block; + use crate::eth::types::BlockNumber; use crate::eth::types::Hash; + use crate::eth::types::SlotIndex; + use crate::eth::types::SlotValue; + use crate::eth::types::UnixTime; fn codec(block_hash: &str) -> BlockHashCursor { - BlockHashCursor { block_hash: block_hash.parse().unwrap() } + BlockHashCursor { + block_hash: block_hash.parse().unwrap(), + } } + // ------------------------------------------------------------------------- + // Cursor codec tests + // ------------------------------------------------------------------------- + #[test] fn cursor_roundtrip_preserves_hash_and_index() { let original = codec("0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c"); @@ -258,4 +299,190 @@ mod tests { fn hash_parses_for_test_fixture() { let _: Hash = "0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c".parse().unwrap(); } + + // ImporterPageRequest::limit() + + #[test] + fn limit_none_returns_default() { + let req = ImporterPageRequest { cursor: None, limit: None }; + assert_eq!(req.limit(), IMPORTER_PAGE_LIMIT_DEFAULT); + } + + #[test] + fn limit_zero_returns_default() { + let req = ImporterPageRequest { cursor: None, limit: Some(0) }; + assert_eq!(req.limit(), IMPORTER_PAGE_LIMIT_DEFAULT); + } + + #[test] + fn limit_small_value_passed_through() { + let req = ImporterPageRequest { cursor: None, limit: Some(50) }; + assert_eq!(req.limit(), 50); + } + + #[test] + fn limit_large_value_clamped_to_max() { + let req = ImporterPageRequest { + cursor: None, + limit: Some(10_000), + }; + assert_eq!(req.limit(), IMPORTER_PAGE_LIMIT_MAX); + } + + // ImporterPagination::from_params + + #[test] + fn from_params_no_pagination_param_returns_none() { + let params = Params::new(Some("[]")); + let result = ImporterPagination::from_params(params.sequence(), BlockFilter::Latest).expect("ok"); + assert!(result.is_none()); + } + + #[test] + fn from_params_with_cursor_resolves_hash_filter_and_start() { + let block_hash = "0x3355a48e6b3e3a3c9e9c4b3a3f3e3d3c3b3a393837363534333231302f2e2d2c"; + let cursor = codec(block_hash).encode_cursor(5); + let json = format!(r#"[{{"cursor":"{cursor}","limit":10}}]"#); + + let params = Params::new(Some(&json)); + let (filter, pagination) = ImporterPagination::from_params(params.sequence(), BlockFilter::Latest) + .expect("ok") + .expect("pagination present"); + + assert!(matches!(filter, BlockFilter::Hash(h) if h == block_hash.parse::().unwrap())); + assert_eq!(pagination.start, 5); + } + + // block_and_receipts_response (server slicing) + + fn block_with_txs(count: usize) -> Block { + let mut block = Block::new(BlockNumber::from(1u64), UnixTime::from(0u64)); + block.transactions = std::iter::repeat_with(|| Faker.fake()).take(count).collect(); + block + } + + #[test] + fn block_and_receipts_single_page_returns_all() { + let block = block_with_txs(3); + let pagination = ImporterPagination::for_test(0, 10); + let response = pagination.block_and_receipts_response(block).expect("ok"); + + assert_eq!(response.pagination.returned, 3); + assert_eq!(response.pagination.total, 3); + assert_eq!(response.receipts.len(), 3); + assert!(response.pagination.next_cursor.is_none()); + } + + #[test] + fn block_and_receipts_multi_page_slices_correctly() { + let block = block_with_txs(5); + + // page 1: start=0, limit=3 + let pagination = ImporterPagination::for_test(0, 3); + let response = pagination.block_and_receipts_response(block.clone()).expect("ok"); + + assert_eq!(response.pagination.returned, 3); + assert_eq!(response.pagination.total, 5); + assert_eq!(response.receipts.len(), 3); + let cursor = response.pagination.next_cursor.expect("more pages"); + + // decode cursor -> start=3 + let (_, start) = BlockHashCursor::decode_cursor(&cursor).expect("valid cursor"); + assert_eq!(start, 3); + + // page 2: start=3, limit=3 + let pagination = ImporterPagination::for_test(start, 3); + let response = pagination.block_and_receipts_response(block).expect("ok"); + + assert_eq!(response.pagination.returned, 2); + assert_eq!(response.pagination.total, 5); + assert_eq!(response.receipts.len(), 2); + assert!(response.pagination.next_cursor.is_none()); + } + + // block_with_changes_response (3-section slicing) + + fn changes_fixture() -> BlockChangesRocksdb { + let mut account_changes = HashMap::with_hasher(HashBuildHasher::default()); + account_changes.insert(AddressRocksdb([0x01; 20]), AccountChangesRocksdb::default()); + account_changes.insert(AddressRocksdb([0x02; 20]), AccountChangesRocksdb::default()); + account_changes.insert(AddressRocksdb([0x03; 20]), AccountChangesRocksdb::default()); + + let mut slot_changes = HashMap::with_hasher(HashBuildHasher::default()); + slot_changes.insert( + (AddressRocksdb([0x01; 20]), SlotIndexRocksdb::from(SlotIndex::from([0u64, 0, 0, 1]))), + SlotValueRocksdb::from(SlotValue::from([0u64, 0, 0, 1])), + ); + slot_changes.insert( + (AddressRocksdb([0x01; 20]), SlotIndexRocksdb::from(SlotIndex::from([0u64, 0, 0, 2]))), + SlotValueRocksdb::from(SlotValue::from([0u64, 0, 0, 2])), + ); + + BlockChangesRocksdb { account_changes, slot_changes } + } + + fn block_rocksdb_with_txs(count: usize) -> BlockRocksdb { + let mut block = Block::new(BlockNumber::from(1u64), UnixTime::from(0u64)); + block.transactions = std::iter::repeat_with(|| Faker.fake()).take(count).collect(); + BlockRocksdb::from(block) + } + + #[test] + fn block_with_changes_multi_page_slices_three_sections() { + let block = block_rocksdb_with_txs(2); + let changes = changes_fixture(); + // total = 2 txs + 3 accounts + 2 slots = 7 + + // page 1: start=0, limit=3 -> 2 txs + 1 account + let pagination = ImporterPagination::for_test(0, 3); + let response = pagination.block_with_changes_response(block.clone(), changes.clone()).expect("ok"); + + assert_eq!(response.pagination.returned, 3); + assert_eq!(response.pagination.total, 7); + assert_eq!(response.block.transactions.len(), 2); + assert_eq!(response.changes.account_changes.len(), 1); + assert_eq!(response.changes.slot_changes.len(), 0); + let cursor = response.pagination.next_cursor.expect("more pages"); + let (_, start) = BlockHashCursor::decode_cursor(&cursor).expect("valid cursor"); + assert_eq!(start, 3); + + // page 2: start=3, limit=3 -> 2 accounts + 1 slot + let pagination = ImporterPagination::for_test(3, 3); + let response = pagination.block_with_changes_response(block.clone(), changes.clone()).expect("ok"); + + assert_eq!(response.pagination.returned, 3); + assert_eq!(response.block.transactions.len(), 0); + assert_eq!(response.changes.account_changes.len(), 2); + assert_eq!(response.changes.slot_changes.len(), 1); + let cursor = response.pagination.next_cursor.expect("more pages"); + let (_, start) = BlockHashCursor::decode_cursor(&cursor).expect("valid cursor"); + assert_eq!(start, 6); + + // page 3: start=6, limit=3 -> 1 slot + let pagination = ImporterPagination::for_test(6, 3); + let response = pagination.block_with_changes_response(block, changes).expect("ok"); + + assert_eq!(response.pagination.returned, 1); + assert_eq!(response.block.transactions.len(), 0); + assert_eq!(response.changes.account_changes.len(), 0); + assert_eq!(response.changes.slot_changes.len(), 1); + assert!(response.pagination.next_cursor.is_none()); + } + + #[test] + fn block_with_changes_single_page_returns_all() { + let block = block_rocksdb_with_txs(1); + let changes = changes_fixture(); + // total = 1 + 3 + 2 = 6 + + let pagination = ImporterPagination::for_test(0, 10); + let response = pagination.block_with_changes_response(block, changes).expect("ok"); + + assert_eq!(response.pagination.returned, 6); + assert_eq!(response.pagination.total, 6); + assert_eq!(response.block.transactions.len(), 1); + assert_eq!(response.changes.account_changes.len(), 3); + assert_eq!(response.changes.slot_changes.len(), 2); + assert!(response.pagination.next_cursor.is_none()); + } } diff --git a/src/eth/rpc/types/pagination.rs b/src/eth/rpc/types/pagination.rs index e5aa71c29..f76310626 100644 --- a/src/eth/rpc/types/pagination.rs +++ b/src/eth/rpc/types/pagination.rs @@ -202,7 +202,15 @@ mod tests { fn empty_total_is_valid() { let p = paginator(0, 0, 10); let (_, info) = take_all(p, &[]); - assert_eq!(info, CursorPageInfo { limit: 10, returned: 0, total: 0, next_cursor: None }); + assert_eq!( + info, + CursorPageInfo { + limit: 10, + returned: 0, + total: 0, + next_cursor: None + } + ); } #[test] @@ -274,4 +282,125 @@ mod tests { assert_eq!(info.returned, 2); assert_eq!(info.next_cursor, Some("2".to_string())); } + + // PaginatedPageFetcher::collect tests + + use super::PageReducer; + use super::PaginatedPageFetcher; + + /// Stub page carrying data and the next cursor (None = last page). + #[derive(Clone)] + struct StubPage { + data: Vec, + next_cursor: Option, + } + + /// Stub reducer that accumulates `StubPage` data into a single concatenated vec. + struct CountingReducer { + pages: Vec>, + } + + impl PageReducer for CountingReducer { + type Output = Vec; + type NextPage = String; + + fn reduce(&mut self, page: StubPage) -> anyhow::Result> { + self.pages.push(page.data); + Ok(page.next_cursor) + } + + fn finish_after_not_found(self) -> anyhow::Result> { + if self.pages.is_empty() { + Ok(None) + } else { + Err(anyhow::anyhow!("block disappeared while fetching paginated pages")) + } + } + + fn finish(self) -> anyhow::Result> { + let merged = self.pages.into_iter().flatten().collect(); + Ok(Some(merged)) + } + } + + #[tokio::test] + async fn collect_happy_path_two_pages() { + let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() }); + let pages = [ + StubPage { + data: vec![1, 2, 3], + next_cursor: Some("page-2".to_string()), + }, + StubPage { + data: vec![4, 5], + next_cursor: None, + }, + ]; + let mut call_idx = 0; + let result = fetcher + .collect(|_cursor: Option| { + let idx = call_idx; + call_idx += 1; + let page = pages.get(idx).cloned(); + async move { Ok(page) } + }) + .await + .expect("ok"); + + let merged = result.expect("some output"); + assert_eq!(merged, vec![1, 2, 3, 4, 5]); + } + + #[tokio::test] + async fn collect_not_found_immediately_returns_none() { + let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() }); + let result = fetcher.collect(|_cursor: Option| async { Ok(None) }).await.expect("ok"); + assert!(result.is_none()); + } + + #[tokio::test] + async fn collect_not_found_after_partial_errors() { + let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() }); + let mut call_idx = 0; + let result = fetcher + .collect(|_cursor: Option| { + let idx = call_idx; + call_idx += 1; + async move { + if idx == 0 { + Ok(Some(StubPage { + data: vec![1, 2, 3], + next_cursor: Some("page-2".to_string()), + })) + } else { + Ok(None) + } + } + }) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn collect_error_propagates() { + let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() }); + let result = fetcher.collect(|_cursor: Option| async { Err(anyhow::anyhow!("network error")) }).await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn collect_single_page_no_cursor() { + let fetcher = PaginatedPageFetcher::new(CountingReducer { pages: Vec::new() }); + let result = fetcher + .collect(|_cursor: Option| async { + Ok(Some(StubPage { + data: vec![42], + next_cursor: None, + })) + }) + .await + .expect("ok"); + let merged = result.expect("some output"); + assert_eq!(merged, vec![42]); + } } diff --git a/src/eth/types/external/external_block.rs b/src/eth/types/external/external_block.rs index 71f844723..3c37e75c0 100644 --- a/src/eth/types/external/external_block.rs +++ b/src/eth/types/external/external_block.rs @@ -161,3 +161,107 @@ impl TryFrom for ExternalBlock { } } } + +#[cfg(test)] +mod tests { + use alloy_primitives::B256; + use alloy_rpc_types_eth::BlockTransactions; + use fake::Fake; + use fake::Faker; + + use super::ExternalBlock; + use crate::eth::types::ExternalTransaction; + + // Builds an ExternalBlock with a fixed hash and `count` random full transactions. + fn block_with_txs(count: usize) -> ExternalBlock { + let mut block: ExternalBlock = Faker.fake(); + block.0.header.hash = fixed_hash(); + let txs: Vec = std::iter::repeat_with(|| Faker.fake()).take(count).collect(); + block.0.transactions = BlockTransactions::Full(txs); + block + } + + fn fixed_hash() -> B256 { + B256::from_slice(&[0xAA; 32]) + } + + fn as_full(block: &ExternalBlock) -> &Vec { + let BlockTransactions::Full(txs) = &block.0.transactions else { + unreachable!("expected full transactions"); + }; + txs + } + + #[test] + fn full_transactions_len_with_full() { + let block = block_with_txs(3); + assert_eq!(block.full_transactions_len().expect("full transactions"), 3); + } + + #[test] + fn full_transactions_len_with_hashes() { + let mut block = block_with_txs(0); + block.0.transactions = BlockTransactions::Hashes(vec![fixed_hash()]); + assert!(block.full_transactions_len().is_err()); + } + + #[test] + fn full_transactions_len_with_uncle() { + let mut block = block_with_txs(0); + block.0.transactions = BlockTransactions::Uncle; + assert!(block.full_transactions_len().is_err()); + } + + #[test] + fn extend_full_transactions_from_same_hash_merges() { + let mut target = block_with_txs(2); + let other = block_with_txs(1); + + target.extend_full_transactions_from(other).expect("same hash merges"); + + assert_eq!(target.full_transactions_len().expect("full transactions"), 3); + } + + #[test] + fn extend_full_transactions_from_different_hash_errors() { + let mut target = block_with_txs(1); + let mut other = block_with_txs(1); + other.0.header.hash = B256::from_slice(&[0xBB; 32]); + + assert!(target.extend_full_transactions_from(other).is_err()); + } + + #[test] + fn extend_full_transactions_from_non_full_source_errors() { + let mut target = block_with_txs(1); + let mut other = block_with_txs(0); + other.0.transactions = BlockTransactions::Hashes(vec![fixed_hash()]); + + assert!(target.extend_full_transactions_from(other).is_err()); + } + + #[test] + fn extend_full_transactions_into_non_full_target_errors() { + let mut target = block_with_txs(0); + target.0.transactions = BlockTransactions::Hashes(vec![fixed_hash()]); + let other = block_with_txs(1); + + assert!(target.extend_full_transactions_from(other).is_err()); + } + + #[test] + fn extend_full_transactions_preserves_order() { + let mut target = block_with_txs(1); + let original = as_full(&target).clone(); + let other = block_with_txs(2); + let other_txs = as_full(&other).clone(); + + target.extend_full_transactions_from(other).expect("same hash merges"); + + let merged = as_full(&target); + assert_eq!(merged.len(), 3); + assert_eq!(merged[0], original[0]); + assert_eq!(merged[1], other_txs[0]); + assert_eq!(merged[2], other_txs[1]); + } +}