refactor(torii-indexer): single fetch range but chunk block processing #2899
Conversation
WalkthroughOhayo, sensei! The changes in the Changes
Sequence DiagramsequenceDiagram
participant Indexer as Engine
participant BlockChain
Indexer->>BlockChain: fetch_data(cursors)
BlockChain-->>Indexer: Return latest block number
Indexer->>Indexer: get_contract_ranges(cursors, latest_block)
Indexer->>BlockChain: fetch_range(contract_ranges)
BlockChain-->>Indexer: Return event data
Indexer->>Indexer: Process event data
The sequence diagram illustrates the high-level flow of data fetching and processing within the indexer's engine, showcasing the interactions between the 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
Documentation and Community
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
crates/torii/indexer/src/engine.rs (1)
328-334: Ohayo sensei! Let's simplify the logging.The log message contains redundant information by repeating the same values twice.
- info!(target: LOG_TARGET, - from = %cursors.head.unwrap_or(self.config.world_block), - to = %latest_block.block_number, - "Fetching data from {} to {}", - cursors.head.unwrap_or(self.config.world_block), - latest_block.block_number - ); + info!(target: LOG_TARGET, + from = %cursors.head.unwrap_or(self.config.world_block), + to = %latest_block.block_number, + "Fetching data" + );
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(2 hunks)
🧰 Additional context used
🪛 GitHub Actions: ci
crates/torii/indexer/src/engine.rs
[warning] 337-337: Formatting issue: Line needs to be split into multiple lines for better readability
[warning] 344-344: Formatting issue: Line needs to be split into multiple lines for better readability
[warning] 555-555: Formatting issue: Extra whitespace needs to be removed
[warning] 562-562: Formatting issue: Extra whitespace needs to be removed
[warning] 603-603: Formatting issue: Line needs to be split into multiple lines for better readability
🔇 Additional comments (3)
crates/torii/indexer/src/engine.rs (3)
337-353: LGTM! Clean and well-structured fetch logic.The implementation handles edge cases appropriately and includes helpful debug timing information.
🧰 Tools
🪛 GitHub Actions: ci
[warning] 337-337: Formatting issue: Line needs to be split into multiple lines for better readability
[warning] 344-344: Formatting issue: Line needs to be split into multiple lines for better readability
555-604: Excellent chunked processing implementation, sensei!The chunked processing with proper block tracking and transaction handling looks solid.
🧰 Tools
🪛 GitHub Actions: ci
[warning] 555-555: Formatting issue: Extra whitespace needs to be removed
[warning] 562-562: Formatting issue: Extra whitespace needs to be removed
[warning] 603-603: Formatting issue: Line needs to be split into multiple lines for better readability
563-563: Consider validating the chunk size.The chunk size should be validated to ensure it's greater than zero to prevent potential infinite loops.
Add validation in the EngineConfig implementation:
impl EngineConfig { pub fn new(blocks_chunk_size: u64) -> Result<Self> { if blocks_chunk_size == 0 { return Err(anyhow!("blocks_chunk_size must be greater than zero")); } Ok(Self { blocks_chunk_size, ..Default::default() }) } }
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
crates/torii/indexer/src/engine.rs (2)
327-347: Ohayo sensei! Fix formatting issues in fetch_data method.The logic looks good, but there are some formatting issues to address:
- Remove extra whitespace at line 327
- Split long lines at 331-332 and 337-338 for better readability
- info!(target: LOG_TARGET, from = %from, to = %latest_block.block_number, "Retrieving block events."); + info!( + target: LOG_TARGET, + from = %from, + to = %latest_block.block_number, + "Retrieving block events." + ); - info!(target: LOG_TARGET, latest_block_number = %latest_block.block_number, "Retrieving pending block."); + info!( + target: LOG_TARGET, + latest_block_number = %latest_block.block_number, + "Retrieving pending block." + );🧰 Tools
🪛 GitHub Actions: ci
[warning] 331-332: Line too long, needs to be split across multiple lines for better formatting
[warning] 337-338: Line too long, needs to be split across multiple lines for better formatting
549-598: Fix formatting issues in process_range method.Please address the following formatting issues:
- Remove extra whitespace at lines 549 and 556
- Split the long line at 597-598
- current_block = chunk_end + 1; + current_block = chunk_end + 1;🧰 Tools
🪛 GitHub Actions: ci
[warning] 549-549: Extra whitespace line detected
[warning] 556-556: Extra whitespace line detected
[warning] 597-598: Line too long, needs to be split across multiple lines for better formatting
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(2 hunks)
🧰 Additional context used
🪛 GitHub Actions: ci
crates/torii/indexer/src/engine.rs
[warning] 324-324: Extra whitespace line detected
[warning] 331-332: Line too long, needs to be split across multiple lines for better formatting
[warning] 337-338: Line too long, needs to be split across multiple lines for better formatting
[warning] 549-549: Extra whitespace line detected
[warning] 556-556: Extra whitespace line detected
[warning] 597-598: Line too long, needs to be split across multiple lines for better formatting
🔇 Additional comments (2)
crates/torii/indexer/src/engine.rs (2)
549-598: Excellent chunking implementation for block processing!The refactoring to process transactions in chunks based on block numbers is a great optimization. It provides better control over memory usage and allows for more efficient processing of large block ranges.
🧰 Tools
🪛 GitHub Actions: ci
[warning] 549-549: Extra whitespace line detected
[warning] 556-556: Extra whitespace line detected
[warning] 597-598: Line too long, needs to be split across multiple lines for better formatting
Line range hint
327-600: Ohayo sensei! Overall excellent refactoring work!The changes demonstrate a thoughtful approach to improving the indexer's performance:
- Better control over block processing through chunking
- Comprehensive logging for debugging
- Well-implemented parallelization
- Proper error handling throughout
Keep up the great work! 🚀
🧰 Tools
🪛 GitHub Actions: ci
[warning] 549-549: Extra whitespace line detected
[warning] 556-556: Extra whitespace line detected
[warning] 597-598: Line too long, needs to be split across multiple lines for better formatting
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #2899 +/- ##
==========================================
+ Coverage 55.82% 55.85% +0.03%
==========================================
Files 449 449
Lines 57693 57732 +39
==========================================
+ Hits 32206 32246 +40
+ Misses 25487 25486 -1 ☔ View full report in Codecov by Sentry. |
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
crates/torii/indexer/src/engine.rs (3)
368-391: Consider parallelizing first event block retrieval.The method currently processes contracts sequentially. For better performance, consider using
futures::stream::iter(self.contracts.iter()).map().buffer_unordered()to parallelize the first event block retrieval.async fn get_contract_ranges( &self, cursors: &Cursors, latest_block: u64, ) -> Result<HashMap<Felt, ContractRange>> { let mut ranges = HashMap::new(); let base_from = cursors.head.unwrap_or(self.config.world_block); - - for (contract_address, _) in self.contracts.iter() { + use futures::stream::{StreamExt, TryStreamExt}; + let contract_ranges = futures::stream::iter(self.contracts.iter()) + .map(|(contract_address, _)| async move { + let first_event_block = + self.get_first_event_block(base_from, latest_block, *contract_address).await?; + Ok::<_, anyhow::Error>((*contract_address, first_event_block)) + }) + .buffer_unordered(10) + .try_collect::<Vec<_>>() + .await?; + + for (contract_address, first_event_block) in contract_ranges { // First fetch a single chunk to find the first event block - let first_event_block = - self.get_first_event_block(base_from, latest_block, *contract_address).await?; - if let Some(first_block) = first_event_block { let from = if first_block == 0 { first_block } else { first_block + 1 }; let total_remaining = latest_block - from; let blocks_to_process = total_remaining.min(self.config.blocks_chunk_size); let to = from + blocks_to_process; - ranges.insert(*contract_address, ContractRange { from, to }); + ranges.insert(contract_address, ContractRange { from, to }); } } Ok(ranges)
Line range hint
413-486: Add context to errors for better debugging.While the concurrent processing is well-implemented, consider adding more context to errors using
anyhow::Context. This will help in debugging issues across multiple contract ranges.async fn fetch_range( &mut self, contract_ranges: HashMap<Felt, ContractRange>, ) -> Result<FetchRangeResult> { let mut fetch_all_events_tasks = VecDeque::new(); for (contract_address, range) in contract_ranges.iter() { - if let Some(contract_type) = self.contracts.get(contract_address) { + if let Some(contract_type) = self.contracts.get(contract_address).with_context(|| format!("Contract {:#x} not found", contract_address))? { let events_filter = EventFilter { from_block: Some(BlockId::Number(range.from)), to_block: Some(BlockId::Number(range.to)), address: Some(*contract_address), keys: None, }; let token_events_pages = - get_all_events(&self.provider, events_filter, self.config.events_chunk_size); + get_all_events(&self.provider, events_filter, self.config.events_chunk_size) + .with_context(|| format!("Failed to fetch events for contract {:#x}", contract_address));
Line range hint
574-611: Enhance observability with structured logging.Consider adding more structured logging to track the progress of range processing, especially for large ranges.
async fn process_range(&mut self, data: FetchRangeResult) -> Result<()> { + debug!( + target: LOG_TARGET, + total_transactions = data.transactions.len(), + "Starting range processing" + ); let mut processed_blocks = HashSet::new(); let mut cursor_map = HashMap::new(); for ((block_number, transaction_hash), events) in data.transactions { - debug!("Processing transaction hash: {:#x}", transaction_hash); + debug!( + target: LOG_TARGET, + %block_number, + %transaction_hash, + events_count = events.len(), + "Processing transaction" + );
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(7 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (4)
- GitHub Check: ensure-wasm
- GitHub Check: build
- GitHub Check: clippy
- GitHub Check: docs
🔇 Additional comments (3)
crates/torii/indexer/src/engine.rs (3)
210-214: LGTM! Clean and focused struct definition.The
ContractRangestruct is well-designed for its purpose of representing block ranges for contract events.
330-366: Ohayo, sensei! Nice refactor of the fetch_data method!The changes improve efficiency by introducing chunked processing and better separation of concerns between range and pending block handling.
394-411: Efficient implementation of first event block retrieval!The method is well-optimized by using a limit of 1 event and properly handling the case when no events are found.
Summary by CodeRabbit
Improvements
Bug Fixes
New Features