feat(torii-indexer): parallelize models & event messages #2912
Conversation
WalkthroughOhayo, sensei! The pull request introduces substantial modifications to the task management system in the Changes
Possibly related PRs
Suggested Reviewers
Sensei, the changes look quite intriguing! The transition to a 🪧 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 (2)
crates/torii/indexer/src/engine.rs (2)
881-896: Document priority levels and consider using constants, sensei!The task identifier generation logic cleverly uses bit manipulation for priorities, but could benefit from better documentation and constant definitions:
+ // Priority levels for different event types + const PRIORITY_MODEL_EVENT: u64 = 0; + const PRIORITY_STORE_OPS: u64 = 2; + // Mask to clear priority bits + const PRIORITY_MASK: u64 = 0x00FFFFFFFFFFFFFF; + match processor.event_key().as_str() { "ModelRegistered" | "EventRegistered" => { let mut hasher = DefaultHasher::new(); event.keys[1].hash(&mut hasher); - // Priority 0 (highest) for model/event registration - let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; - hash | (0u64 << 56) + // Use highest 8 bits for priority, lower 56 bits for hash-based grouping + let hash = hasher.finish() & PRIORITY_MASK; + hash | (PRIORITY_MODEL_EVENT << 56) }
601-657: Enhance error handling with more context, sensei!The task processing implementation looks solid, but could benefit from improved error handling. Consider adding more context to errors and enhancing debug logging:
if let Err(e) = processor .process( &world, &mut local_db, event.block_number, event.block_timestamp, &event.event_id, &event.event, &event_processor_config, ) .await { error!( target: LOG_TARGET, event_name = processor.event_key(), + block_number = %event.block_number, + timestamp = %event.block_timestamp, error = %e, task_id = %task_id, + "Failed to process event. This might impact dependent operations.", - "Processing parallelized event." ); + // Propagate error with additional context + return Err(anyhow::anyhow!("Failed to process event {}: {}", event.event_id, e)); }
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(4 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: fmt
🔇 Additional comments (1)
crates/torii/indexer/src/engine.rs (1)
219-219: Ohayo! Nice choice using BTreeMap for ordered task processing, sensei!The switch from
HashMaptoBTreeMapenables natural priority ordering of tasks, which is crucial for maintaining the correct processing sequence of model registrations and store operations.Also applies to: 253-253
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
crates/torii/indexer/src/engine.rs (2)
256-256: Ohayo, sensei! Confirm consistent initialization oftasks.Initializing
taskswithBTreeMap::new()is appropriate. However, if there are default priorities or initial tasks, consider initializing them here for clarity.
603-666: Ohayo, sensei! Potential task starvation due to sequential priority processing.Processing tasks strictly sequentially by priority level may cause lower-priority tasks to be delayed indefinitely if higher-priority tasks are continuously added. Consider implementing a more dynamic scheduling strategy to balance task priorities and prevent starvation.
Do you want assistance in proposing an enhanced task scheduling mechanism?
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(5 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: fmt
🔇 Additional comments (3)
crates/torii/indexer/src/engine.rs (3)
210-212: Ohayo, sensei! Good use of type aliases for clarity.Introducing
TaskPriorityandTaskIdimproves code readability and ensures type safety.
222-222: Ohayo, sensei! Verify the implications of changingtasksto aBTreeMap.Switching from
HashMaptoBTreeMapfortasksintroduces ordered task management based on priority. Ensure this change aligns with the desired task processing logic and does not introduce unintended side effects.
889-903:⚠️ Potential issueOhayo, sensei! Potential hash collisions in
task_identifiergeneration.Using truncated hash values for
task_identifiermay lead to collisions, causing unrelated tasks to share the same identifier. Consider using a more collision-resistant method or including additional unique data to ensure task identifiers are unique.Please run the following script to check for hash collisions:
crates/torii/indexer/src/engine.rs
Outdated
| self.tasks | ||
| .entry(task_priority) | ||
| .or_default() | ||
| .entry(task_identifier) | ||
| .or_default() | ||
| .push(( | ||
| contract_type, | ||
| ParallelizedEvent { | ||
| event_id: event_id.to_string(), | ||
| event: event.clone(), | ||
| block_number, | ||
| block_timestamp, | ||
| }, | ||
| )); | ||
| } else { | ||
| // Process non-parallelized events immediately |
There was a problem hiding this comment.
Ohayo, sensei! Review handling of task_identifier equal to zero.
When task_identifier is zero, events are processed immediately outside of the priority queue. Ensure that zero cannot be a valid task_identifier that should be queued, or adjust the logic to handle zero appropriately to prevent unintended bypassing of the task queue.
crates/torii/indexer/src/engine.rs
Outdated
| let (task_priority, task_identifier) = match processor.event_key().as_str() { | ||
| "ModelRegistered" | "EventRegistered" => { | ||
| let mut hasher = DefaultHasher::new(); | ||
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | ||
| let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; | ||
| (0usize, hash) // Priority 0 (highest) for model/event registration | ||
| } | ||
| "StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { | ||
| let mut hasher = DefaultHasher::new(); | ||
| // model selector | ||
| event.keys[1].hash(&mut hasher); | ||
| // entity id | ||
| event.keys[2].hash(&mut hasher); | ||
| hasher.finish() | ||
| let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; | ||
| (2usize, hash) // Priority 2 (lower) for store operations | ||
| } | ||
| _ => 0, | ||
| _ => (0, 0) // No parallelization for other events |
There was a problem hiding this comment.
Ohayo, sensei! Guard against index out-of-bounds when accessing event.keys.
Accessing event.keys[1] and event.keys[2] without checking their existence may lead to panics if the keys vector is shorter than expected. Ensure that the event.keys vector has the required length before accessing these indices.
Apply this diff to add necessary checks:
+ if event.keys.len() >= 3 {
let mut hasher = DefaultHasher::new();
event.keys[1].hash(&mut hasher);
event.keys[2].hash(&mut hasher);
let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF;
(2usize, hash) // Priority 2 (lower) for store operations
+ } else {
+ warn!(target: LOG_TARGET, "Insufficient event keys for hashing. Using default task identifier.");
+ (2usize, 0) // Handle with default or error as appropriate
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let (task_priority, task_identifier) = match processor.event_key().as_str() { | |
| "ModelRegistered" | "EventRegistered" => { | |
| let mut hasher = DefaultHasher::new(); | |
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | |
| let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; | |
| (0usize, hash) // Priority 0 (highest) for model/event registration | |
| } | |
| "StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { | |
| let mut hasher = DefaultHasher::new(); | |
| // model selector | |
| event.keys[1].hash(&mut hasher); | |
| // entity id | |
| event.keys[2].hash(&mut hasher); | |
| hasher.finish() | |
| let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; | |
| (2usize, hash) // Priority 2 (lower) for store operations | |
| } | |
| _ => 0, | |
| _ => (0, 0) // No parallelization for other events | |
| let (task_priority, task_identifier) = match processor.event_key().as_str() { | |
| "ModelRegistered" | "EventRegistered" => { | |
| let mut hasher = DefaultHasher::new(); | |
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | |
| let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; | |
| (0usize, hash) // Priority 0 (highest) for model/event registration | |
| } | |
| "StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { | |
| if event.keys.len() >= 3 { | |
| let mut hasher = DefaultHasher::new(); | |
| event.keys[1].hash(&mut hasher); | |
| event.keys[2].hash(&mut hasher); | |
| let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; | |
| (2usize, hash) // Priority 2 (lower) for store operations | |
| } else { | |
| warn!(target: LOG_TARGET, "Insufficient event keys for hashing. Using default task identifier."); | |
| (2usize, 0) // Handle with default or error as appropriate | |
| } | |
| } | |
| _ => (0, 0) // No parallelization for other events |
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
crates/torii/indexer/src/engine.rs (1)
895-901:⚠️ Potential issueAdd safety check for event keys access.
Guard against potential index out-of-bounds when accessing
event.keys."StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { + if event.keys.len() < 3 { + warn!( + target: LOG_TARGET, + "Insufficient event keys for hashing. Using default task identifier." + ); + return (2usize, 0); + } let mut hasher = DefaultHasher::new(); event.keys[1].hash(&mut hasher); event.keys[2].hash(&mut hasher); let hash = hasher.finish(); (2usize, hash) }🧰 Tools
🪛 GitHub Actions: ci
[warning] 900-900: Missing comma in match expression
🧹 Nitpick comments (6)
crates/torii/indexer/src/processors/store_del_record.rs (1)
66-72: Excellent error message improvement, but there's a small formatting issue, sensei!The enhanced error message with hex-formatted selector will make debugging much easier. However, there's a formatting issue to fix.
Fix the extra closing parenthesis on line 68 as flagged by the CI:
- return Err(anyhow::anyhow!( - "Failed to retrieve model with selector {:#x}: {}", - event.selector, - e - )) + return Err(anyhow::anyhow!( + "Failed to retrieve model with selector {:#x}: {}", + event.selector, + e + ));🧰 Tools
🪛 GitHub Actions: ci
[warning] 68-68: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_update_record.rs (1)
71-75: Ohayo! Small formatting fix needed, sensei!The CI pipeline indicates an extra closing parenthesis. Please fix the formatting to pass the pipeline checks.
- return Err(anyhow::anyhow!( - "Failed to retrieve model with selector {:#x}: {}", - event.selector, - e - )) + return Err(anyhow::anyhow!("Failed to retrieve model with selector {:#x}: {}", event.selector, e))🧰 Tools
🪛 GitHub Actions: ci
[warning] 72-72: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_set_record.rs (1)
67-73: Fix formatting and enhance error context, sensei!The error handling provides good context, but there are two improvements to consider:
- Fix the formatting issue with the extra closing parenthesis.
- Consider including the model namespace/name in the error message for better debugging.
Apply this diff:
- Err(e) => { - return Err(anyhow::anyhow!( - "Failed to retrieve model with selector {:#x}: {}", - event.selector, - e - )) - } + Err(e) => return Err(anyhow::anyhow!( + "Failed to retrieve model with selector {:#x} (namespace: {}, name: {}): {}", + event.selector, + model.namespace, + model.name, + e + ))🧰 Tools
🪛 GitHub Actions: ci
[warning] 69-69: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_update_member.rs (1)
64-71: Enhanced error handling with debug logging, sensei!The addition of debug logging provides better visibility into model absence cases.
Consider using string interpolation for a more idiomatic debug message:
- debug!( - target: LOG_TARGET, - selector = %model_selector, - "Model does not exist, skipping." - ); + debug!( + target: LOG_TARGET, + "Model with selector {model_selector:#x} does not exist, skipping." + );crates/torii/indexer/src/engine.rs (2)
600-662: Clean up formatting issues.Please address the following formatting issues:
- Remove extra whitespace lines at 600, 607, and 616
- Split the chain method calls at lines 623-625 for better readability
- let processor = processors.iter() - .find(|p| p.validate(&event.event)) - .expect("Must find at least one processor for the event"); + let processor = processors + .iter() + .find(|p| p.validate(&event.event)) + .expect("Must find at least one processor for the event");🧰 Tools
🪛 GitHub Actions: ci
[warning] 600-600: Extra whitespace line found, formatting issue
[warning] 607-607: Extra whitespace line found, formatting issue
[warning] 616-616: Extra whitespace line found, formatting issue
[warning] 623-625: Chain method calls should be split into multiple lines for better readability
907-920: Format method chain for better readability.Split the long chain of method calls across multiple lines.
- self.tasks - .entry(task_priority) - .or_default() - .entry(task_identifier) - .or_default() - .push(( - contract_type, - ParallelizedEvent { - event_id: event_id.to_string(), - event: event.clone(), - block_number, - block_timestamp, - }, - )); + self.tasks + .entry(task_priority) + .or_default() + .entry(task_identifier) + .or_default() + .push(( + contract_type, + ParallelizedEvent { + event_id: event_id.to_string(), + event: event.clone(), + block_number, + block_timestamp, + }, + ));🧰 Tools
🪛 GitHub Actions: ci
[warning] 907-917: Long chain of method calls should be formatted across multiple lines
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
crates/torii/indexer/src/engine.rs(5 hunks)crates/torii/indexer/src/processors/store_del_record.rs(2 hunks)crates/torii/indexer/src/processors/store_set_record.rs(2 hunks)crates/torii/indexer/src/processors/store_update_member.rs(3 hunks)crates/torii/indexer/src/processors/store_update_record.rs(2 hunks)
🧰 Additional context used
🪛 GitHub Actions: ci
crates/torii/indexer/src/processors/store_set_record.rs
[warning] 69-69: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_update_member.rs
[warning] 74-74: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_del_record.rs
[warning] 68-68: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_update_record.rs
[warning] 72-72: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/engine.rs
[warning] 600-600: Extra whitespace line found, formatting issue
[warning] 607-607: Extra whitespace line found, formatting issue
[warning] 616-616: Extra whitespace line found, formatting issue
[warning] 623-625: Chain method calls should be split into multiple lines for better readability
[warning] 900-900: Missing comma in match expression
[warning] 907-917: Long chain of method calls should be formatted across multiple lines
🔇 Additional comments (11)
crates/torii/indexer/src/processors/store_del_record.rs (2)
38-38: Ohayo! Parameter renaming looks good, sensei!Removing the underscore prefix is appropriate since we're now using the config parameter.
58-65: Nice error handling improvement, sensei!The additional check for non-empty namespaces prevents silently ignoring errors when no namespaces are configured, making the behavior more explicit and safer.
crates/torii/indexer/src/processors/store_update_record.rs (2)
39-39: Ohayo! Parameter renaming looks good, sensei!The change from
_configtoconfigproperly reflects that this parameter is now being utilized in the implementation.
62-76: Excellent error handling improvements, sensei!The changes enhance error handling in two valuable ways:
- The condition
!config.namespaces.is_empty()ensures we only skip missing models when explicitly filtering by namespaces- The error message now includes the selector in hex format for better debugging
🧰 Tools
🪛 GitHub Actions: ci
[warning] 72-72: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/processors/store_set_record.rs (2)
39-39: Ohayo! Parameter rename looks good, sensei!The removal of the underscore prefix is appropriate since the
configparameter is now actively used in the error handling logic.
59-66: Ohayo! Nice enhancement to the namespace-specific indexing logic, sensei!The condition
!config.namespaces.is_empty()effectively handles the case where only specific namespaces are being indexed.crates/torii/indexer/src/processors/store_update_member.rs (3)
10-10: Ohayo! Import changes look good, sensei!The addition of
debugto thetracingimport aligns well with the new debug logging capability.
40-40: Parameter rename reflects its usage, sensei!The rename from
_configtoconfigaccurately reflects that the parameter is now actively used in the function.
73-77:⚠️ Potential issueFix the formatting issue, sensei!
There's an extra closing parenthesis as indicated by the pipeline failure.
Apply this fix:
- return Err(anyhow::anyhow!( - "Failed to retrieve model with selector {:#x}: {}", - event.selector, - e - )) + return Err(anyhow::anyhow!( + "Failed to retrieve model with selector {:#x}: {}", + event.selector, + e + ))Likely invalid or redundant comment.
🧰 Tools
🪛 GitHub Actions: ci
[warning] 74-74: Extra closing parenthesis, formatting issue
crates/torii/indexer/src/engine.rs (2)
Line range hint
210-256: Ohayo, sensei! LGTM - Clean type definitions and struct changes.The introduction of type aliases and the switch to
BTreeMapfor tasks improves code organization and enables priority-based processing.
603-666: Ohayo, sensei! Excellent implementation of priority-based task processing.The sequential processing of priority levels while maintaining concurrency within each level is well-designed.
🧰 Tools
🪛 GitHub Actions: ci
[warning] 607-607: Extra whitespace line found, formatting issue
[warning] 616-616: Extra whitespace line found, formatting issue
[warning] 623-625: Chain method calls should be split into multiple lines for better readability
crates/torii/indexer/src/engine.rs
Outdated
| let (task_priority, task_identifier) = match processor.event_key().as_str() { | ||
| "ModelRegistered" | "EventRegistered" => { | ||
| let mut hasher = DefaultHasher::new(); | ||
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | ||
| let hash = hasher.finish(); | ||
| (0usize, hash) // Priority 0 (highest) for model/event registration | ||
| } | ||
| "StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { | ||
| let mut hasher = DefaultHasher::new(); | ||
| // model selector | ||
| event.keys[1].hash(&mut hasher); | ||
| // entity id | ||
| event.keys[2].hash(&mut hasher); | ||
| hasher.finish() | ||
| let hash = hasher.finish(); | ||
| (2usize, hash) // Priority 2 (lower) for store operations | ||
| } | ||
| _ => 0, | ||
| _ => (0, 0) // No parallelization for other events |
There was a problem hiding this comment.
Fix match expression syntax.
Add missing comma in the match expression.
"StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => {
let mut hasher = DefaultHasher::new();
event.keys[1].hash(&mut hasher);
event.keys[2].hash(&mut hasher);
let hash = hasher.finish();
- (2usize, hash) // Priority 2 (lower) for store operations
+ (2usize, hash), // Priority 2 (lower) for store operations
}
_ => (0, 0) // No parallelization for other events📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let (task_priority, task_identifier) = match processor.event_key().as_str() { | |
| "ModelRegistered" | "EventRegistered" => { | |
| let mut hasher = DefaultHasher::new(); | |
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | |
| let hash = hasher.finish(); | |
| (0usize, hash) // Priority 0 (highest) for model/event registration | |
| } | |
| "StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { | |
| let mut hasher = DefaultHasher::new(); | |
| // model selector | |
| event.keys[1].hash(&mut hasher); | |
| // entity id | |
| event.keys[2].hash(&mut hasher); | |
| hasher.finish() | |
| let hash = hasher.finish(); | |
| (2usize, hash) // Priority 2 (lower) for store operations | |
| } | |
| _ => 0, | |
| _ => (0, 0) // No parallelization for other events | |
| let (task_priority, task_identifier) = match processor.event_key().as_str() { | |
| "ModelRegistered" | "EventRegistered" => { | |
| let mut hasher = DefaultHasher::new(); | |
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | |
| let hash = hasher.finish(); | |
| (0usize, hash) // Priority 0 (highest) for model/event registration | |
| } | |
| "StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { | |
| let mut hasher = DefaultHasher::new(); | |
| event.keys[1].hash(&mut hasher); | |
| event.keys[2].hash(&mut hasher); | |
| let hash = hasher.finish(); | |
| (2usize, hash), // Priority 2 (lower) for store operations | |
| } | |
| _ => (0, 0) // No parallelization for other events |
🧰 Tools
🪛 GitHub Actions: ci
[warning] 900-900: Missing comma in match expression
Larkooo
changed the title
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
crates/torii/indexer/src/engine.rs (2)
605-668: Ohayo, sensei! Consider enhancing error handling in task processing.While the priority-based processing logic is solid, the error handling could be more robust. Currently, errors in individual event processing are logged but don't affect other events or tasks.
Consider implementing a more sophisticated error handling strategy:
for (contract_type, event) in events { let contract_processors = processors.get_event_processor(contract_type); if let Some(processors) = contract_processors.get(&event.event.keys[0]) { - let processor = processors - .iter() - .find(|p| p.validate(&event.event)) - .expect("Must find at least one processor for the event"); + let processor = processors + .iter() + .find(|p| p.validate(&event.event)) + .ok_or_else(|| anyhow::anyhow!("No valid processor found for event"))?; debug!( target: LOG_TARGET, event_name = processor.event_key(), task_id = %task_id, priority = %priority, "Processing parallelized event." ); - if let Err(e) = processor - .process( + processor + .process( &world, &mut local_db, event.block_number, event.block_timestamp, &event.event_id, &event.event, &event_processor_config, ) - .await - { - error!( - target: LOG_TARGET, - event_name = processor.event_key(), - error = %e, - task_id = %task_id, - priority = %priority, - "Processing parallelized event." - ); - } + .await + .map_err(|e| { + error!( + target: LOG_TARGET, + event_name = processor.event_key(), + error = %e, + task_id = %task_id, + priority = %priority, + "Processing parallelized event." + ); + e + })?; } }
837-837: Ohayo, sensei! Enhance error context in raw event storage.Consider adding more context to the error when storing raw events fails.
-self.db.store_event(event_id, event, transaction_hash, block_timestamp)?; +self.db.store_event(event_id, event, transaction_hash, block_timestamp) + .with_context(|| format!("Failed to store raw event {}", event_id))?;
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(8 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: fmt
🔇 Additional comments (2)
crates/torii/indexer/src/engine.rs (2)
Line range hint
211-223: Ohayo, sensei! Excellent choice of data structures for task prioritization.The use of
BTreeMapfor the tasks field ensures natural ordering by priority, which is essential for sequential processing of priority levels. The type aliases improve code readability.
884-928:⚠️ Potential issueOhayo, sensei! Guard against array access in event key handling.
When processing store operations, the code assumes
event.keys[1]andevent.keys[2]exist. This could lead to panics if the event keys array is shorter than expected.Add bounds checking:
"StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { + if event.keys.len() < 3 { + error!( + target: LOG_TARGET, + "Insufficient event keys for store operation. Expected at least 3 keys." + ); + return Ok(()); + } let mut hasher = DefaultHasher::new(); event.keys[1].hash(&mut hasher); event.keys[2].hash(&mut hasher); let hash = hasher.finish(); (2usize, hash) // Priority 2 (lower) for store operations }Likely invalid or redundant comment.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
crates/torii/indexer/src/engine.rs (1)
605-668: Ohayo, sensei! Consider enhancing task processing robustness.While the implementation effectively manages task prioritization and concurrency, consider these improvements:
- Add timeout handling for task groups to prevent priority level deadlocks
- Implement more granular error handling for individual task failures
for (priority, task_group) in std::mem::take(&mut self.tasks) { + let timeout = Duration::from_secs(30); // Configurable timeout let mut handles = Vec::new(); // Process all tasks within this priority level concurrently for (task_id, events) in task_group { let db = self.db.clone(); let world = self.world.clone(); let semaphore = semaphore.clone(); let processors = self.processors.clone(); let event_processor_config = self.config.event_processor_config.clone(); handles.push(tokio::spawn(async move { let _permit = semaphore.acquire().await?; let mut local_db = db.clone(); + let mut task_errors = Vec::new(); // Process all events for this task sequentially for (contract_type, event) in events { let contract_processors = processors.get_event_processor(contract_type); if let Some(processors) = contract_processors.get(&event.event.keys[0]) { let processor = processors .iter() .find(|p| p.validate(&event.event)) .expect("Must find at least one processor for the event"); if let Err(e) = processor .process( &world, &mut local_db, event.block_number, event.block_timestamp, &event.event_id, &event.event, &event_processor_config, ) .await { - error!( + task_errors.push(( target: LOG_TARGET, event_name = processor.event_key(), error = %e, task_id = %task_id, priority = %priority, "Processing parallelized event." - ); + )); } } } + if !task_errors.is_empty() { + error!(target: LOG_TARGET, "Task {task_id} completed with {} errors", task_errors.len()); + for (event_name, error, task_id, priority) in task_errors { + error!(target: LOG_TARGET, %event_name, %error, %task_id, %priority, "Event processing error"); + } + } Ok::<_, anyhow::Error>(()) })); } // Wait for all tasks in this priority level to complete before moving to next priority - try_join_all(handles).await?; + match tokio::time::timeout(timeout, try_join_all(handles)).await { + Ok(result) => result?, + Err(_) => { + error!(target: LOG_TARGET, "Priority level {priority} tasks timed out"); + return Err(anyhow::anyhow!("Task processing timeout")); + } + } }
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
crates/torii/indexer/src/engine.rs(8 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: fmt
🔇 Additional comments (2)
crates/torii/indexer/src/engine.rs (2)
211-212: Ohayo, sensei! LGTM - Well-structured task prioritization system.The type aliases and BTreeMap implementation provide a clear and ordered approach to task prioritization.
Also applies to: 223-223, 257-257
884-914:⚠️ Potential issueOhayo, sensei! Guard against array access and hash collisions.
The task identifier generation has potential issues:
- Missing bounds check when accessing
event.keys[1]andevent.keys[2]- Hash collisions could lead to task merging
Apply this diff to add necessary checks:
"StoreSetRecord" | "StoreUpdateRecord" | "StoreUpdateMember" | "StoreDelRecord" => { + if event.keys.len() < 3 { + error!(target: LOG_TARGET, "Insufficient keys for store operation event"); + return (0, 0); // Process immediately as fallback + } let mut hasher = DefaultHasher::new(); event.keys[1].hash(&mut hasher); event.keys[2].hash(&mut hasher); - let hash = hasher.finish(); + // Use more bits to reduce collision probability + let hash = hasher.finish() & 0x00FFFFFFFFFFFFFF; (2usize, hash) // Priority 2 (lower) for store operations }Likely invalid or redundant comment.
| "EventEmitted" => { | ||
| let mut hasher = DefaultHasher::new(); | ||
|
|
||
| let keys = Vec::<Felt>::cairo_deserialize(&event.data, 0).unwrap_or_else(|e| { | ||
| panic!("Expected EventEmitted keys to be well formed: {:?}", e); | ||
| }); | ||
|
|
||
| // selector | ||
| event.keys[1].hash(&mut hasher); | ||
| // entity id | ||
| let entity_id = poseidon_hash_many(&keys); | ||
| entity_id.hash(&mut hasher); | ||
|
|
||
| let hash = hasher.finish(); | ||
| (2usize, hash) // Priority 2 for event messages | ||
| } |
There was a problem hiding this comment.
Ohayo, sensei! Replace panic with proper error handling in EventEmitted processing.
The current implementation panics on deserialization failure, which is not production-safe.
Apply this diff to handle errors gracefully:
"EventEmitted" => {
let mut hasher = DefaultHasher::new();
- let keys = Vec::<Felt>::cairo_deserialize(&event.data, 0).unwrap_or_else(|e| {
- panic!("Expected EventEmitted keys to be well formed: {:?}", e);
- });
+ let keys = match Vec::<Felt>::cairo_deserialize(&event.data, 0) {
+ Ok(keys) => keys,
+ Err(e) => {
+ error!(
+ target: LOG_TARGET,
+ error = ?e,
+ "Failed to deserialize EventEmitted keys. Processing immediately."
+ );
+ return (0, 0); // Process immediately as fallback
+ }
+ };
// selector
event.keys[1].hash(&mut hasher);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| "EventEmitted" => { | |
| let mut hasher = DefaultHasher::new(); | |
| let keys = Vec::<Felt>::cairo_deserialize(&event.data, 0).unwrap_or_else(|e| { | |
| panic!("Expected EventEmitted keys to be well formed: {:?}", e); | |
| }); | |
| // selector | |
| event.keys[1].hash(&mut hasher); | |
| // entity id | |
| let entity_id = poseidon_hash_many(&keys); | |
| entity_id.hash(&mut hasher); | |
| let hash = hasher.finish(); | |
| (2usize, hash) // Priority 2 for event messages | |
| } | |
| "EventEmitted" => { | |
| let mut hasher = DefaultHasher::new(); | |
| let keys = match Vec::<Felt>::cairo_deserialize(&event.data, 0) { | |
| Ok(keys) => keys, | |
| Err(e) => { | |
| error!( | |
| target: LOG_TARGET, | |
| error = ?e, | |
| "Failed to deserialize EventEmitted keys. Processing immediately." | |
| ); | |
| return (0, 0); // Process immediately as fallback | |
| } | |
| }; | |
| // selector | |
| event.keys[1].hash(&mut hasher); | |
| // entity id | |
| let entity_id = poseidon_hash_many(&keys); | |
| entity_id.hash(&mut hasher); | |
| let hash = hasher.finish(); | |
| (2usize, hash) // Priority 2 for event messages | |
| } |
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #2912 +/- ##
==========================================
- Coverage 55.22% 55.20% -0.02%
==========================================
Files 449 449
Lines 58335 58395 +60
==========================================
+ Hits 32215 32237 +22
- Misses 26120 26158 +38 ☔ View full report in Codecov by Sentry. |
| @@ -207,6 +208,9 @@ pub struct ParallelizedEvent { | |||
| pub event: Event, | |||
| } | |||
|
|
|||
| type TaskPriority = usize; | |||
There was a problem hiding this comment.
Wdyt about an enum, using the Ord trait to keep keys ordered in the BTreeMap?
Feeling like enum would be more idiomatic, and remove the magic values like we have in the engine.
There was a problem hiding this comment.
Something like:
enum TaskPriority {
Low,
Medium,
High,
}| let mut hasher = DefaultHasher::new(); | ||
| event.keys.iter().for_each(|k| k.hash(&mut hasher)); | ||
| let hash = hasher.finish(); | ||
| (0usize, hash) // Priority 0 (highest) for model/event registration |
There was a problem hiding this comment.
| (0usize, hash) // Priority 0 (highest) for model/event registration | |
| (TaskPriority::High, hash) // Priority 0 (highest) for model/event registration |
| event.keys[2].hash(&mut hasher); | ||
| hasher.finish() | ||
| let hash = hasher.finish(); | ||
| (2usize, hash) // Priority 2 (lower) for store operations |
There was a problem hiding this comment.
| (2usize, hash) // Priority 2 (lower) for store operations | |
| (TaskPriority::Low, hash) // Priority 2 (lower) for store operations |
Summary by CodeRabbit