opt(torii-core): move off queryqueue for executing tx

bin/torii/src/main.rs

@@ -31,6 +31,7 @@ use tokio::sync::broadcast;
 use tokio::sync::broadcast::Sender;
 use tokio_stream::StreamExt;
 use torii_core::engine::{Engine, EngineConfig, IndexingFlags, Processors};
+use torii_core::executor::Executor;
 use torii_core::processors::event_message::EventMessageProcessor;
 use torii_core::processors::generate_event_processors_map;
 use torii_core::processors::metadata_update::MetadataUpdateProcessor;
@@ -185,7 +186,8 @@ async fn main() -> anyhow::Result<()> {
     // Get world address
     let world = WorldContractReader::new(args.world_address, provider.clone());
 
-    let db = Sql::new(pool.clone(), args.world_address).await?;
+    let (mut executor, sender) = Executor::new(pool.clone(), shutdown_tx.clone()).await?;
+    let db = Sql::new(pool.clone(), args.world_address, sender.clone()).await?;
 
     let processors = Processors {
         event: generate_event_processors_map(vec![
@@ -289,6 +291,7 @@ async fn main() -> anyhow::Result<()> {
 
     tokio::select! {
         res = engine.start() => res?,
+        _ = executor.run() => {},
         _ = proxy_server.start(shutdown_tx.subscribe()) => {},
         _ = graphql_server => {},
         _ = grpc_server => {},

crates/torii/core/src/engine.rs

@@ -7,6 +7,7 @@ use std::time::Duration;
 use anyhow::Result;
 use bitflags::bitflags;
 use dojo_world::contracts::world::WorldContractReader;
+use futures_util::future::try_join_all;
 use hashlink::LinkedHashMap;
 use starknet::core::types::{
     BlockId, BlockTag, EmittedEvent, Event, EventFilter, Felt, MaybePendingBlockWithReceipts,
@@ -17,7 +18,6 @@ use starknet::providers::Provider;
 use tokio::sync::broadcast::Sender;
 use tokio::sync::mpsc::Sender as BoundedSender;
 use tokio::sync::Semaphore;
-use tokio::task::JoinSet;
 use tokio::time::{sleep, Instant};
 use tracing::{debug, error, info, trace, warn};
 
@@ -151,7 +151,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
         // use the start block provided by user if head is 0
         let (head, _, _) = self.db.head().await?;
         if head == 0 {
-            self.db.set_head(self.config.start_block);
+            self.db.set_head(self.config.start_block)?;
         } else if self.config.start_block != 0 {
             warn!(target: LOG_TARGET, "Start block ignored, stored head exists and will be used instead.");
         }
@@ -179,7 +179,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                             }
 
                             match self.process(fetch_result).await {
-                                Ok(()) => {}
+                                Ok(()) => self.db.execute()?,
                                 Err(e) => {
                                     error!(target: LOG_TARGET, error = %e, "Processing fetched data.");
                                     erroring_out = true;
@@ -407,15 +407,14 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                             // provider. So we can fail silently and try
                             // again in the next iteration.
                             warn!(target: LOG_TARGET, transaction_hash = %format!("{:#x}", transaction_hash), "Retrieving pending transaction receipt.");
-                            self.db.set_head(data.block_number - 1);
+                            self.db.set_head(data.block_number - 1)?;
                             if let Some(tx) = last_pending_block_tx {
-                                self.db.set_last_pending_block_tx(Some(tx));
+                                self.db.set_last_pending_block_tx(Some(tx))?;
                             }
 
                             if let Some(tx) = last_pending_block_world_tx {
-                                self.db.set_last_pending_block_world_tx(Some(tx));
+                                self.db.set_last_pending_block_world_tx(Some(tx))?;
                             }
-                            self.db.execute().await?;
                             return Ok(());
                         }
                         _ => {
@@ -441,18 +440,16 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
 
         // Set the head to the last processed pending transaction
         // Head block number should still be latest block number
-        self.db.set_head(data.block_number - 1);
+        self.db.set_head(data.block_number - 1)?;
 
         if let Some(tx) = last_pending_block_tx {
-            self.db.set_last_pending_block_tx(Some(tx));
+            self.db.set_last_pending_block_tx(Some(tx))?;
         }
 
         if let Some(tx) = last_pending_block_world_tx {
-            self.db.set_last_pending_block_world_tx(Some(tx));
+            self.db.set_last_pending_block_world_tx(Some(tx))?;
         }
 
-        self.db.execute().await?;
-
         Ok(())
     }
 
@@ -486,20 +483,14 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                 self.process_block(block_number, data.blocks[&block_number]).await?;
                 last_block = block_number;
             }
-
-            if self.db.query_queue.queue.len() >= QUERY_QUEUE_BATCH_SIZE {
-                self.db.execute().await?;
-            }
         }
 
         // Process parallelized events
         self.process_tasks().await?;
 
-        self.db.set_head(data.latest_block_number);
-        self.db.set_last_pending_block_world_tx(None);
-        self.db.set_last_pending_block_tx(None);
-
-        self.db.execute().await?;
+        self.db.set_head(data.latest_block_number)?;
+        self.db.set_last_pending_block_world_tx(None)?;
+        self.db.set_last_pending_block_tx(None)?;
 
         Ok(())
     }
@@ -509,14 +500,14 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
         let semaphore = Arc::new(Semaphore::new(self.config.max_concurrent_tasks));
 
         // Run all tasks concurrently
-        let mut set = JoinSet::new();
+        let mut handles = Vec::new();
         for (task_id, events) in self.tasks.drain() {
             let db = self.db.clone();
             let world = self.world.clone();
             let processors = self.processors.clone();
             let semaphore = semaphore.clone();
 
-            set.spawn(async move {
+            handles.push(tokio::spawn(async move {
                 let _permit = semaphore.acquire().await.unwrap();
                 let mut local_db = db.clone();
                 for ParallelizedEvent { event_id, event, block_number, block_timestamp } in events {
@@ -532,14 +523,11 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                     }
                 }
                 Ok::<_, anyhow::Error>(local_db)
-            });
+            }));
         }
 
         // Join all tasks
-        while let Some(result) = set.join_next().await {
-            let local_db = result??;
-            self.db.merge(local_db)?;
-        }
+        try_join_all(handles).await?;
 
         Ok(())
     }
@@ -688,7 +676,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
         transaction_hash: Felt,
     ) -> Result<()> {
         if self.config.flags.contains(IndexingFlags::RAW_EVENTS) {
-            self.db.store_event(event_id, event, transaction_hash, block_timestamp);
+            self.db.store_event(event_id, event, transaction_hash, block_timestamp)?;
         }
 
         let event_key = event.keys[0];