feat(torii-indexer): configurable chunked batch requests (#37)

* feat(torii-indexer): configurable chunked batch requests

* better error

* fmt

* c

Author: Larkooo

crates/cli/src/options.rs

@@ -16,6 +16,7 @@ pub const DEFAULT_METRICS_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
 pub const DEFAULT_METRICS_PORT: u16 = 9200;
 pub const DEFAULT_EVENTS_CHUNK_SIZE: u64 = 1024;
 pub const DEFAULT_BLOCKS_CHUNK_SIZE: u64 = 10240;
+pub const DEFAULT_BATCH_CHUNK_SIZE: usize = 1024;
 pub const DEFAULT_POLLING_INTERVAL: u64 = 500;
 pub const DEFAULT_MAX_CONCURRENT_TASKS: usize = 100;
 pub const DEFAULT_RELAY_PORT: u16 = 9090;
@@ -189,13 +190,22 @@ pub struct IndexingOptions {
         help = "Whether or not to read models from the block number they were registered in."
     )]
     pub strict_model_reader: bool,
+
+    /// The chunk size to use for batch requests.
+    #[arg(
+        long = "indexing.batch_chunk_size",
+        default_value_t = DEFAULT_BATCH_CHUNK_SIZE,
+        help = "The chunk size to use for batch requests. This is used to split the requests into smaller chunks to avoid overwhelming the provider and potentially running into issues."
+    )]
+    pub batch_chunk_size: usize,
 }
 
 impl Default for IndexingOptions {
     fn default() -> Self {
         Self {
             events_chunk_size: DEFAULT_EVENTS_CHUNK_SIZE,
             blocks_chunk_size: DEFAULT_BLOCKS_CHUNK_SIZE,
+            batch_chunk_size: DEFAULT_BATCH_CHUNK_SIZE,
             pending: true,
             polling_interval: DEFAULT_POLLING_INTERVAL,
             max_concurrent_tasks: DEFAULT_MAX_CONCURRENT_TASKS,

crates/indexer/src/engine.rs

@@ -4,10 +4,11 @@ use std::hash::Hash;
 use std::sync::Arc;
 use std::time::Duration;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use bitflags::bitflags;
 use dojo_utils::provider as provider_utils;
 use dojo_world::contracts::world::WorldContractReader;
+use futures_util::future::try_join_all;
 use hashlink::LinkedHashMap;
 use starknet::core::types::requests::{
     GetBlockWithTxHashesRequest, GetEventsRequest, GetTransactionByHashRequest,
@@ -42,6 +43,7 @@ bitflags! {
 #[derive(Debug)]
 pub struct EngineConfig {
     pub polling_interval: Duration,
+    pub batch_chunk_size: usize,
     pub blocks_chunk_size: u64,
     pub events_chunk_size: u64,
     pub max_concurrent_tasks: usize,
@@ -54,6 +56,7 @@ impl Default for EngineConfig {
     fn default() -> Self {
         Self {
             polling_interval: Duration::from_millis(500),
+            batch_chunk_size: 1024,
             blocks_chunk_size: 10240,
             events_chunk_size: 1024,
             max_concurrent_tasks: 100,
@@ -349,7 +352,8 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                 }
             }
 
-            let transaction_results = self.provider.batch_requests(transaction_requests).await?;
+            let transaction_results = self.chunked_batch_requests(&transaction_requests).await?;
+
             for (block_number, result) in block_numbers.into_iter().zip(transaction_results) {
                 match result {
                     ProviderResponseData::GetTransactionByHash(transaction) => {
@@ -382,7 +386,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
 
         // Execute timestamp requests in batch
         if !timestamp_requests.is_empty() {
-            let timestamp_results = self.provider.batch_requests(timestamp_requests).await?;
+            let timestamp_results = self.chunked_batch_requests(&timestamp_requests).await?;
 
             // Process timestamp results
             for (block_number, result) in block_numbers.iter().zip(timestamp_results) {
@@ -426,7 +430,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                 .iter()
                 .map(|(_, req)| req.clone())
                 .collect();
-            let batch_results = self.provider.batch_requests(batch_requests).await?;
+            let batch_results = self.chunked_batch_requests(&batch_requests).await?;
 
             // Process results and prepare next batch of requests if needed
             for ((contract_address, original_request), result) in
@@ -898,6 +902,34 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
 
         Ok(())
     }
+
+    async fn chunked_batch_requests(
+        &self,
+        requests: &[ProviderRequestData],
+    ) -> Result<Vec<ProviderResponseData>> {
+        if requests.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        let mut futures = Vec::new();
+        for chunk in requests.chunks(self.config.batch_chunk_size) {
+            futures.push(async move { self.provider.batch_requests(chunk).await });
+        }
+
+        let results_of_chunks: Vec<Vec<ProviderResponseData>> = try_join_all(futures)
+            .await
+            .with_context(|| {
+                format!(
+                    "One or more batch requests failed during chunked execution. This could be due to the provider being overloaded. You can try reducing the batch chunk size. Total requests: {}. Batch chunk size: {}",
+                    requests.len(),
+                    self.config.batch_chunk_size
+                )
+            })?;
+
+        let flattened_results = results_of_chunks.into_iter().flatten().collect();
+
+        Ok(flattened_results)
+    }
 }
 
 // event_id format: block_number:transaction_hash:event_idx

crates/runner/src/lib.rs

@@ -278,6 +278,7 @@ impl Runner {
                 max_concurrent_tasks: self.args.indexing.max_concurrent_tasks,
                 blocks_chunk_size: self.args.indexing.blocks_chunk_size,
                 events_chunk_size: self.args.indexing.events_chunk_size,
+                batch_chunk_size: self.args.indexing.batch_chunk_size,
                 polling_interval: Duration::from_millis(self.args.indexing.polling_interval),
                 flags,
                 event_processor_config: EventProcessorConfig {